카프카 메시지 consume은 대략 다음과 같은 절차로 진행된다.

1. Fetch Serialized Data from Broker(직렬화된 데이터를 브로커로부터 가져와서)
2. Deserialize the fetched serialized data(정해진 타입으로 역직렬화하고)
3. Consume the deserialized data(역직렬화한 데이터를 처리하고)
4. Commit(브로커에게 commit 신호 전송)

받아온 메시지 처리에 실패하면..

스프링 카프카를 사용하면서 개발자가 @KafkaListener를 붙여 작성한 리스너 구현부는 3번 과정에 해당한다.
3번 과정에서 에러가 발생하면 정해진 횟수(기본 10회)만큼 재시도 후 끝까지 실패하면 대략 아래와 같은 에러 로그를 남기면서 그냥 4번 commit 신호를 브로커로 보낸다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

2022-08-27 22:50:06.326 ERROR 64718 --- [ntainer#0-0-C-1] o.s.kafka.listener.DefaultErrorHandler   : Backoff FixedBackOff{interval=0, currentAttempts=10, maxAttempts=9} exhausted for basic-topic-01-1@2

org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method 'public void io.homo_efficio.scratchpad.spring.reactor.kafka.service.KafkaConsumer.listenerBasicTopic01(java.lang.String)' threw exception; nested exception is java.lang.RuntimeException: 4th 메시지는 처리 불가!!!; nested exception is java.lang.RuntimeException: 4th 메시지는 처리 불가!!!
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.decorateException(KafkaMessageListenerContainer.java:2703) ~[spring-kafka-2.8.7.jar:2.8.7]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeOnMessage(KafkaMessageListenerContainer.java:2673) ~[spring-kafka-2.8.7.jar:2.8.7]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeOnMessage(KafkaMessageListenerContainer.java:2633) ~[spring-kafka-2.8.7.jar:2.8.7]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:2560) ~[spring-kafka-2.8.7.jar:2.8.7]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeWithRecords(KafkaMessageListenerContainer.java:2441) ~[spring-kafka-2.8.7.jar:2.8.7]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:2319) ~[spring-kafka-2.8.7.jar:2.8.7]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:1990) ~[spring-kafka-2.8.7.jar:2.8.7]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeIfHaveRecords(KafkaMessageListenerContainer.java:1366) ~[spring-kafka-2.8.7.jar:2.8.7]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1357) ~[spring-kafka-2.8.7.jar:2.8.7]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1252) ~[spring-kafka-2.8.7.jar:2.8.7]
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) ~[na:na]
    at java.base/java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:264) ~[na:na]
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java) ~[na:na]
    at java.base/java.lang.Thread.run(Thread.java:832) ~[na:na]
    Suppressed: org.springframework.kafka.listener.ListenerExecutionFailedException: Restored Stack Trace
        at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:363) ~[spring-kafka-2.8.7.jar:2.8.7]
        at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:92) ~[spring-kafka-2.8.7.jar:2.8.7]
        at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:53) ~[spring-kafka-2.8.7.jar:2.8.7]
        at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeOnMessage(KafkaMessageListenerContainer.java:2653) ~[spring-kafka-2.8.7.jar:2.8.7]
Caused by: java.lang.RuntimeException: 4th 메시지는 처리 불가!!!
    at io.homo_efficio.scratchpad.spring.reactor.kafka.service.KafkaConsumer.listenerBasicTopic01(KafkaConsumer.kt:13) ~[main/:na]
    at jdk.internal.reflect.GeneratedMethodAccessor85.invoke(Unknown Source) ~[na:na]
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:na]
    at java.base/java.lang.reflect.Method.invoke(Method.java:564) ~[na:na]
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:169) ~[spring-messaging-5.3.21.jar:5.3.21]
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:119) ~[spring-messaging-5.3.21.jar:5.3.21]
    at org.springframework.kafka.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:56) ~[spring-kafka-2.8.7.jar:2.8.7]
    at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:347) ~[spring-kafka-2.8.7.jar:2.8.7]
    at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:92) ~[spring-kafka-2.8.7.jar:2.8.7]
    at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:53) ~[spring-kafka-2.8.7.jar:2.8.7]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeOnMessage(KafkaMessageListenerContainer.java:2653) ~[spring-kafka-2.8.7.jar:2.8.7]
    ... 12 common frames omitted

브로커는 commit 신호를 받아서 해당 토픽, 파티션의 offset을 증가시킨다.
컨수머 입장에서는 해당 데이터 처리에 실패했지만, 브로커 입장에서는 정상적으로 데이터가 처리됐을 때와 마찬가지로 commit 신호를 받으므로 장애는 발생하지 않는다.

역직렬화에 실패하면..

그런데 2번 역직렬화 과정에서 에러가 나면 어떻게 될까?

예를 들어 JsonSerializer/JsonDeserializer를 사용해서 JSON으로 직렬화/역직렬화하도록 구성했는데,
JSON 규격에 맞지 않는 메시지(poison pill)가 인입되면 역직렬화 과정에서 에러가 발생하게 된다.

2번 역직렬화 과정은 3번 과정보다 앞서 진행되므로 2번 과정에서 에러가 발생하면 개발자가 작성한 리스너로 진입하지 못하며,
그 앞 단계에서 즉 프레임워크 수준에서 에러 처리가 돼야 한다.
그런데 이 에러 처리를 별로 신경쓰지 않고 사용하다가 역직렬화 과정에서 에러가 발생하면,
3번 과정에서 처럼 정해진 횟수만큼 재시도하다가 얌전히 포기하고 commit 처리되는 게 아니라,
역직렬화에 성공할 때까지 계속, 그것도 아주 빠른 속도로 재시도한다.

그런데 역직렬화 실패한 데이터를 다시 역직렬화 하면 성공할 수 있을까?
그럴 가능성은 거의 없다고 생각하는데 안타깝게도 카프카 컨수머는, 아니 최소한 카프카 스프링 컨수머는 역직렬화 안 되면 될 때까지!! 서버 빠개질때까지!! 그렇게 불사파처럼 무대뽀로 동작한다.

그 결과 컨수머 시스템 자원이 갉아먹히는데,
이 컨수머가 다른 토픽의 컨수머이기도 하면 자원 부족으로 그 다른 토픽의 메시지를 컨숨하는 속도가 느려지고 그 다른 토픽에도 메시지가 쌓이게 된다.
결국 이 컨수머를 포함하고 있는 서버 인스턴스들은 점점 좀비화되고 서비스 장애 상황으로 이어진다.

그러니 역직렬화에 실패하면,
마치 혁명에 실패하면 반역죄로 능지처사에 3족이 멸문 당하는 것처럼 아주 걍 작살이 나는 거시다.

Spring ErrorHandlingDeserializer

위에 ‘(역직렬화 과정) 에러 처리를 별로 신경쓰지 않고 기본 설정으로 두고 사용하면’이라는 단서가 있는데,
바꿔말하면 역직렬화 에러 처리를 신경쓰면 위와 같은 장애 발생을 막을 수 있다. 어떻게?

스프링 카프카 2.2에 도입된 ErrorHandlingDeserializer를 사용하고, ErrorHandler를 지정해주면 된다.

이 내용은 역직렬화 재시도를 계속하면서 펑펑 싸대는 아래와 같은 로그에 나름 친절하게 나와있다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

2022-08-27 14:03:22.077 ERROR 81052 --- [ntainer#0-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : Consumer exception

java.lang.IllegalStateException: This error handler cannot process 'SerializationException's directly; please consider configuring an 'ErrorHandlingDeserializer' in the value and/or key deserializer
    at org.springframework.kafka.listener.DefaultErrorHandler.handleOtherException(DefaultErrorHandler.java:149) ~[spring-kafka-2.8.7.jar:2.8.7]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.handleConsumerException(KafkaMessageListenerContainer.java:1799) ~[spring-kafka-2.8.7.jar:2.8.7]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1298) ~[spring-kafka-2.8.7.jar:2.8.7]
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) ~[na:na]
    at java.base/java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:264) ~[na:na]
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java) ~[na:na]
    at java.base/java.lang.Thread.run(Thread.java:832) ~[na:na]
Caused by: org.apache.kafka.common.errors.RecordDeserializationException: Error deserializing key/value for partition basic-topic-01-2 at offset 5. If needed, please seek past the record to continue consumption.
    at org.apache.kafka.clients.consumer.internals.Fetcher.parseRecord(Fetcher.java:1448) ~[kafka-clients-3.1.1.jar:na]
    at org.apache.kafka.clients.consumer.internals.Fetcher.access$3400(Fetcher.java:135) ~[kafka-clients-3.1.1.jar:na]
    at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.fetchRecords(Fetcher.java:1671) ~[kafka-clients-3.1.1.jar:na]
    at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.access$1900(Fetcher.java:1507) ~[kafka-clients-3.1.1.jar:na]
    at org.apache.kafka.clients.consumer.internals.Fetcher.fetchRecords(Fetcher.java:733) ~[kafka-clients-3.1.1.jar:na]
    at org.apache.kafka.clients.consumer.internals.Fetcher.fetchedRecords(Fetcher.java:684) ~[kafka-clients-3.1.1.jar:na]
    at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1277) ~[kafka-clients-3.1.1.jar:na]
    at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1238) ~[kafka-clients-3.1.1.jar:na]
    at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1211) ~[kafka-clients-3.1.1.jar:na]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollConsumer(KafkaMessageListenerContainer.java:1522) ~[spring-kafka-2.8.7.jar:2.8.7]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doPoll(KafkaMessageListenerContainer.java:1512) ~[spring-kafka-2.8.7.jar:2.8.7]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1340) ~[spring-kafka-2.8.7.jar:2.8.7]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1252) ~[spring-kafka-2.8.7.jar:2.8.7]
    ... 4 common frames omitted
Caused by: java.lang.IllegalStateException: No type information in headers and no default type provided
    at org.springframework.util.Assert.state(Assert.java:76) ~[spring-core-5.3.21.jar:5.3.21]
    at org.springframework.kafka.support.serializer.JsonDeserializer.deserialize(JsonDeserializer.java:583) ~[spring-kafka-2.8.7.jar:2.8.7]
    at org.apache.kafka.clients.consumer.internals.Fetcher.parseRecord(Fetcher.java:1439) ~[kafka-clients-3.1.1.jar:na]
    ... 16 common frames omitted

물론 공식 문서에도 나와있다.
(하지만 우리는 에러가 발생하기 전까지는 공식 문서를 잘 안..=3=3)

어쨌든 ErrorHandlingDeserializer 설정 방법이나 알아보자.

ErrorHandlingDeserializer

스프링 카프카 설정 방법은 여러가지인데 결국에는 KafkaConsumerFactory에 ErrorHandlingDeserializer를 지정해주면 된다.

소스 코드로는 대략 다음과 같이 작성하면 되고,

1
2
3
4
5
6
7
8
9
10

// from https://docs.spring.io/spring-kafka/reference/html/#error-handling-deserializer

... // other props
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);

props.put(ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS, JsonDeserializer.class);
props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class.getName());

return DefaultKafkaConsumerFactory<>(props);

yml로 다음과 같이 작성해도 효과는 같다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

spring:  
  kafka:
    consumer:
      bootstrap-servers:
        - localhost:29092
      key-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      properties:
        spring:
          deserializer:
            key:
              delegate:
                class: org.springframework.kafka.support.serializer.JsonDeserializer
            value:
              delegate:
                class: org.springframework.kafka.support.serializer.JsonDeserializer

결국 ErrorHandlingDeserializer로 JsonDeserializer를 감싸서, 에러가 있으면 ErrorHandler에게 넘기고, 에러가 없으면 JsonDeserializer에게 넘기는 구조다.

ErrorHandler

에러 핸들러 지정 방식도 여러가지인데 가장 간단하게는 KafkaListenerContainerFactory에 ErrorHandler를 지정해주면 된다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

@Bean
fun kafkaListenerContainerFactory(): ConcurrentKafkaListenerContainerFactory<String, String> {
    
    val factory = ConcurrentKafkaListenerContainerFactory<String, String>()
    
    // ErrorHandler 지정
    factory.setCommonErrorHandler(                  
        DefaultErrorHandler { record, exception ->
            log.error(
                """
                    Consume 실패!!!
                    cause: ${exception.message}
                    topic: ${record.topic()}
                    partition: ${record.partition()}
                    offset: ${record.offset()}
                    message key: ${record.key()}
                    message value: ${record.value()}
                    ${exception.stackTraceToString()}
                """.trimIndent()
            )
        }
    )

    factory.consumerFactory = consumerFactory()  // 앞서 ErrorHandlingDeserializer 설정할 때 만든 DefaultKafkaConsumerFactory

    return factory
}

단순히 로깅만으로 끝내도 되면 위와 같이 간단히 처리할 수 있고, 그게 아니라면 dead-letter topic으로 전송되게 할 수도 있다.

dead-letter topic 관련은 https://docs.spring.io/spring-kafka/reference/html/#dead-letters 를 참고하자.

이밖에도 @KafkaListener에서 에러 처리를 지정할 수도 있고,
consumerProps.put(ErrorHandlingDeserializer.VALUE_FUNCTION, FailedFooProvider.class) 이런 식으로 역직렬화에 실패한 메시지 대신에 다른 fallback 메시지를 만들어서 리스너에게 전달해줄 수도 있다.

마무리

• 카프카 스프링 컨수머에서 역직렬화에 실패하면 완전히 새된다.
• StringDeserializer를 사용하면 역직렬화 실패 가능성은 매우 낮아지지만, 실제 비즈니스에서 사용하려면 역직렬화한 문자열을 다시 업무에 사용하는 데이터 타입으로 변환해줘야 한다.
• 이 과정을 JSON으로 한 번에 해주는 게 JsonDeserializer
• 하지만 JsonDeserializer를 날로 그냥 먹으면 심각한 식중독에 걸릴 수 있다.
• JsonDeserializer 사용 등 역직렬화 오류가 발생할 수 있는 상황에서는 반드시 ErrorHandlingDeserializer를 사용해야 Poison Pill로 인한 장애를 막을 수 있다.

참고

• https://docs.spring.io/spring-kafka/reference/html/#error-handling-deserializer
• https://www.confluent.io/ko-kr/blog/spring-kafka-can-your-kafka-consumers-handle-a-poison-pill/

오징어 전문 쇼핑몰을 만들었는데 입소문이 잘 났다. 그래서 포털 몇 군데에도 API 열어줬더니 대애애애박이 났다. 물 들어 온 김에 노저어야지. 해산물 종합 쇼핑몰로 확장하려고 한다.

그런데 아뿔싸.. 엔티티 이름, 테이블 이름 등이 squid로 돼 있다. 잘못된 건 없다. 애초에 오징어 전문 쇼핑몰이 목표였으니까. 요구사항이 바뀌었고 그에 따라 시스템을 전환해야 하는 일이 생겼을 뿐 잘못된 것은 없다.

어떻게 전환해야 할까? 시스템을 멈추고 점검 화면을 보여주고 전환해도 될 것 같지만, 우리를 돈쭐내고 싶어 안달이난 사용자들을 오랜 시간 대기하게 만드는 것은 예의가 아니다.

시스템 중단 없이 마이그레이션 해보자.

요구사항

• 마이그레이션 전/중/후에 사용자 요청 처리에 아무 오류가 없어야 한다.
• API 클라이언트는 마이그레이션 후에도 당분간 기존 API를 사용할 수 있어야 각자 편한 시기에 새 API로 변경할 수 있어야 한다.
• squid 대신 seafood 엔티티가 중심이 되고 squid는 seafood의 여러 타입 중 하나가 된다.

일반적인 마이그레이션 시나리오

편의상 기존 오징어 전문 쇼핑몰은 OLD 라고 하고, 해산물 종합 쇼핑몰은 NEW 라고 하면 대략 다음과 같은 시나리오로 전환 작업이 진행된다.

1. squid 대신 seafood 기준으로 동작하는 NEW 애플리케이션 작성
2. squid 기준의 OLD db에서 seafood 기준으로 구성된 NEW db로의 데이터 마이그레이션 구성
3. NEW 애플리케이션 배포
4. OLD 애플리케이션으로 향하던 트래픽을 NEW 애플리케이션으로 향하도록 라우팅 변경

Command와 Query의 분리

시스템의 상태를 변경하는 CUD 작업과 상태를 변경하지 않고 R 작업만 수행하는 기능이 하나의 애플리케이션에 들어있다면, 전환도 한 번에 할 수 밖에 없다. 한 번에 전환하는 것이 꼭 나쁘다고 할 수는 없지만, CUD와 R을 분리해서 전환하면 분리 집중, 점진적 실행 등 Divide and Conquer 전략의 일반적인 장점을 누릴 수 있다. 조금 더 구체적으로 기술하면 다음과 같다.

• 상태 변경이 없으므로 전환 작업이 상대적으로 훨씬 수월한 R을 먼저 전환하고 모니터링하면서 seafood 기준의 로직에 오류가 없는지 점검하고 수정할 기회를 가질 수 있다.
• R 전환에서 얻은 경험을 토대로 데이터 일관성 유지에 훨씬 더 많은 주의를 기울어야 할 CUD에 대해 더 면밀히 준비하고, 전환 이후에도 상태 변경에 대해서만 집중적으로 모니터링하고 대처할 수 있다.

따라서 가능하다면 기존 애플리케이션을 CUD를 수행하는 Command와 R을 수행하는 Query로 분리할 수 있다면 분리하는 것이 좋다. 마이그레이션 관점에서는 API Endpoint 수준에서만 분리돼도 충분하다. Endpoint 수준에서의 분리가 말은 쉬워보이지만 다음과 같이 데이터 조회와 수정에 대한 API URL이 동일하게 구성되고 HTTP Method만 다르게 돼 있는 경우에는,

1
2

- 특정 해산물 조회: /api/seafoods/{id} GET
- 특정 해산물 수정: /api/seafoods/{id} PUT

애플리케이션 서버 앞단에 있는 Reverse Proxy에서 API 하나하나마다 HTTP Method 수준의 분기를 구성해야 하므로 굉장히 고된 작업일 수 있다.

따라서 고된 분리 작업을 거쳐서 전환 과정에서의 안정성을 높일지, 아니면 고된 분리 작업을 생략하는 대신 전환 과정에서 발생할 수 있는 위험을 감수하고 한 방에 전환할지 신중하게 고민하고 선택해야 한다.

하지만 API가 애초부터 다음과 같이 구성돼 있었다면 아주 자연스럽게 Command와 Query를 분리 전환할 할 수 있다.

1
2

- 특정 해산물 조회: /query/api/seafoods/{id} GET
- 특정 해산물 수정: /command/api/seafoods/{id} PUT

다행스럽게도 OLD 애플리케이션은 위와 같이 Command와 Query가 분리돼 있었다.

서론이 길었는데 위와 같은 현황과 요구사항을 반영해서 구체화된 설계도는 다음과 같다.

전환 작업 설계

그림에 간략한 설명이 있으니 그냥 주욱 살펴보자.

1. 현황

2. 데이터 마이그레이션

3. NEW Query 애플리케이션 배포

4. Query 라우팅 전환

5. NEW Command 애플리케이션 배포

6. Command 라우팅 전환

7. DB 복제 중지

8. NEW API 라우팅 추가

9. 클라이언트 API 변경

10. OLD 라우팅 제거

11. OLD API Endpoint 제거

12. 마이그레이션 완료

단계별 실제 구현 작업은 다음 편에서 알아보자.

helm은 k8s 패키지 매니저다.

공식 문서에 흔한 그림 하나 없고 이래저래 영 마음에 들지 않아서 개념적으로 이해하는 데 필요 이상의 노력이 드는 것 같아서 어쩔 수 없이 따로 핵심만 추려서 정리해본다.

동작 구조

일단 k8s 패키지 매니저가 뭔지 그림으로 맛을 보자.

이해하기 쉽게 한 가지 방식만을 골라서 단순화 했으며, 실제로는 물론 여러가지 시나리오, 방식으로 구성할 수 있다.

• 개발자가 작성한 Helm Chart 를 helm push 명령으로 Helm Chart Repository에 업로드 한다.
• 개발자가 만든 Container Image 를 docker push 명령으로 Container Registry에 업로드 한다.
• 개발자가 k8s Control Plane 에 values.yaml 파일을 지정하면서 helm install 명령을 전달하면,
  • values.yaml 에 있는 값이 Helm Chart 에 주입되고 Helm Release 가 생성되고,
  • Helm Chart 에 들어있는 Image 정보를 통해 Container Image 를 가져와서 Helm Chart 정보를 토대로 Pod 등 k8s 자원이 생성된다.

주요 용어

helm chart

• k8s 자원 yaml 파일을 만들 수 있는 여러 yaml 템플릿 파일과 설정값이 들어 있는 values.yaml 파일로 이루어진 파일 세트

• 대략 아래와 같은 파일이 모여있는 디렉터리라고 봐도 크게 틀리지 않는다. 아래 내용은 helm 한글 문서 https://helm.sh/ko/docs/topics/charts/#차트-파일-구조 에서 가져왔다.

  1 2 3 4 5 6 7 8 9 10

  CHART이름/ Chart.yaml # 차트에 대한 정보를 가진 YAML 파일 LICENSE # 옵션: 차트의 라이센스 정보를 가진 텍스트 파일 README.md # 옵션: README 파일 values.yaml # 차트에 대한 기본 환경설정 값들 values.schema.json # 옵션: values.yaml 파일의 구조를 제약하는 JSON 파일 charts/ # 이 차트에 종속된 차트들을 포함하는 디렉터리 crds/ # 커스텀 자원에 대한 정의 templates/ # values와 결합될 때, 유효한 쿠버네티스 manifest 파일들이 생성될 템플릿들의 디렉터리 templates/NOTES.txt # 옵션: 간단한 사용법을 포함하는 텍스트 파일

• templates 폴더 안에는 다음과 같은 k8s 자원 yaml 파일이 들어있다. 아래 내용은 helm 한글 문서 https://helm.sh/ko/docs/chart_template_guide/getting_started/#mycharttemplates-훑어보기 에서 가져왔다.

  1 2 3 4

  NOTES.txt : 차트의 "도움말". 이것은 helm install 을 실행할 때 사용자에게 표시될 것이다. deployment.yaml : 쿠버네티스 디플로이먼트를 생성하기 위한 기본 매니페스트 service.yaml : 디플로이먼트의 서비스 엔드포인트를 생성하기 위한 기본 매니페스트 _helpers.tpl : 차트 전체에서 다시 사용할 수 있는 템플릿 헬퍼를 지정하는 공간

  • 예를 들어 deployment.yaml 파일에 아래와 같은 내용이 있고, helm install 실행 시 사용되는 values.yaml 파일에 spring.configLocation이라는 항목(key)이 있다면 그 값을 SPRING_CONFIG_LOCATION 환경변수에 저장한다.

    1 2 3 4 5

    env: {{- if hasKey .Values.spring "configLocation" }} - name: SPRING_CONFIG_LOCATION value: '{{ toString .Values.spring.configLocation }}' {{- end}}

• values.yaml 파일도 포함돼 있는데 이 파일을 사용할 수도 있고, helm chart 밖에 존재하는 별도의 values.yaml 파일을 지정해서 사용할 수도 있다. 위 그림에 나온 방식은 별도의 values.yaml 파일을 사용하고 있다.

helm release

• helm install 명령 실행 결과로 생성되며, helm chart의 인스턴스라고 볼 수 있다.
• 하나의 helm chart 에
  • 서로 다른 values.yaml 을 적용해서 내용적으로 다른 여러 helm release 를 만들 수도 있고,
  • 동일한 values.yaml 을 적용하되 release 이름을 다르게 지정해서 내용적으로는 동일한 여러 helm release 를 만들 수도 있다.
• helm chart 및 values.yaml 파일의 내용 변경을 k8s 자원에 반영해야 할 때 helm upgrade 명령을 통해 release 의 내용을 변경할 수 있다.

주요 명령

사실 이런 건 그냥 공식 문서를 보고 알 수 있어야 하는데, helm 문서에 나온 설명에 혼동을 불러일으키는 부분이 많아서 어쩔 수 없이 일부만 정리한다.

이 명령 사용법만 알면 나머지 다른 명령은 큰 혼동 없이 문서를 보고도 이해할 수 있을 것이다.

helm push

1
2

형식: helm push 올릴chart경로 helmRepo이름
예제: helm push deploy/my-chart my-repo

• git push 와 비슷하다고 보면 된다.
• helm chart를 helm repository에 저장한다.

helm install

1
2

형식: helm install -n k8sNamespace 생성될release이름 사용할chart이름 -f 사용할valuesyaml파일경로
예제: helm install -n my-namespace my-release my-repo/my-chart -f deploy/values-my-values.yaml

• package.json 내용 대로 패키지를 가져와서 설치하는 npm install 과 비슷하다고 보면 된다.
• helm repository에 저장된 helm chart를 가져오고,
• 명시적으로 지정한 values-xxx.yaml 에 있는 값을 helm chart 에 포함돼 있는 여러 k8s yaml template 파일에 주입해서,
• k8s 자원 yaml을 생성할 수 있는 helm release 를 생성하고,
• helm chart 에 지정돼 있는 위치에서 컨테이너 이미지를 가져와서 실제로 k8s에 자원을 생성한다.
• 복잡해 보이지만 위 그림을 다시 보면 더 쉽게 이해할 수 있다.

helm uninstall

1
2

형식: helm uninstall -n k8sNamespace 삭제할release이름
예제: helm uninstall -n my-namespace my-release

• install 에 의해 생성된 helm release를 삭제하고 k8s에 생성됐던 deployment 등 관련 자원도 모두 삭제된다.

helm upgrade

1
2

형식: helm upgrade -n k8sNamespace release이름 사용할chart이름 -f 사용할valuesyaml파일경로
예제: helm upgrade -n my-namespace my-release my-repo/my-chart -f deploy/values-my-other-values.yaml

• install 에 의해 생성된 helm release의 내용을 변경한다.

실무 사용 참고

• 소스 코드에서 컨테이너 이미지를 만들어 컨테이너 레지스트리에 업로드 하는 일은 보통 jenkins 등 CI 도구를 사용해서 처리한다.
• 소스 코드가 공개 repo에 있다면 ArgoCD 를 k8s 클러스터 내부에 구성해서 편리하게 배포할 수 있다.
• 소스 코드가 비공개 repo에 있다면 이 비공개 repo에 접근할 수 있는 곳에 CI용 jenkins를 두고, k8s 클러스터 내부에 CD용 jenkins를 둬서, CI용 jenkins가 이미지를 컨테이너 레지스트리에 올린 후에 CD용 jenkins를 호출(HTTP API)해서 배포할 수 있다.
• 소스 코드가 변경됐다면 그에 따른 image 만 변경하면 되므로 CI-CD만 하면 되고, helm 작업(install 또는 upgrade)은 다시 할 필요가 없다.
• 소스 코드는 변경이 없는데 helm chart나 values.yaml 파일만 변경됐다면 helm upgrade만 다시 하면 배포까지 되고 CI-CD 작업은 다시 할 필요가 없다.

helm chart 작성

작성 방법은 공식 문서 https://helm.sh/ko/docs/chart_template_guide/ 를 참고한다.

비동기 프로그래밍은 늘 어렵다.

이미 오래된 애기지만 Reactive Streams의 구현체인 RxJava나 Reactor가 나오고 Spring에서도 WebFlux가 나오면서 저변이 더욱 확대된 것 같다.
학습에 의해 넘을 수 있다고는 하지만 그것도 일부 잘 하는 개발자들에 대한 얘기같고, 현실적으로는 나같은 일반적인 개발자의 직관에 반하는 부분들이 많아 여전히 어렵고 고통스럽다.
저변이 확대되는 것은 좋지만, 일부 개발자는 어쩌면 겉멋을 부리려고 굳이 쓰지 않아도 되는 곳에 사용하고, 이력서에 Reactor/Reactive 개발 경험을 추가하고 그걸 발판 삼아 회사를 떠난다.
고통은 남아서 이어 받는 사람들의 몫..

자바라면 아직까지는 답이 없고(토비님이 알려주신 https://github.com/vsilaev/tascalate-async-await 같은 서드파티 라이브러리가 있기는 하다),
코틀린이라면 다행히 코루틴(coroutine)이 있다.

둘을 비교해서 코루틴이 쉽고 직관적이라는 자료는 이미 많이 있지만, 그냥 하나 더 덧붙여본다.

Reactor

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

fun fluxRangeSample(): Mono<String> {
    val strList: MutableList<String> = mutableListOf()

    return Flux.range(0, 2)
        .doOnNext { outer ->
            println("outer loop index: $outer")
    
            Flux.range(1, 4)
                .subscribe { inner ->
                    println("  inner loop i: $inner")
                    
                    Mono.just("    DB 호출 없는 Mono: $inner")
                        .subscribe { anyStr -> println(anyStr) }
                    
                    memberRepository.findById("moid117")
                        .subscribe { member ->
                            println("    DB 호출 있는 Mono: ${member.oid} - ${member.name}")
                            strList.add("${member.oid} - ${member.name}")
                        }
                }
        }
        .doOnNext {
            println("Second doOnNext")
        }
        .doOnComplete {
            for (i in 1..5) {
                println("doOnComplete i: $i")
            }
        }
        .then(Mono.just("[${strList.joinToString(" | ")}]"))

사실 보기만 해도 보고 싶은 마음이 별로 들지 않는다. 하지만 봐야 하니 참고 보면, 대충 2중 루프 돌면서 DB에서 읽은 문자열을 strList에 추가해서 반환하는 코드다.

그래서 아래와 같은 컨트롤러를 붙여서 호출하면 |로 구분된 문자열 목록이 나올 것이다.

1
2
3
4
5

@GetMapping("/flux-range")
fun fluxRange(): Mono<ResponseEntity<String>> {
    return memberCommandService.fluxRangeSample()
        .map { ResponseEntity.ok(it) }
}

실행해보면 짠~ 하고

1
2
3
4
5
6

~ 🦑🍕🍺 ❯ http GET :8080/api/members/flux-range
HTTP/1.1 200 OK
Content-Length: 2
Content-Type: text/plain;charset=UTF-8

[]

예상과는 다르게 공허하게 비어있는 리스트가 반환된다. 하지만 로그에 보면 아래와 같이 DB에서 읽은 값도 출력이 된다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

outer loop index: 0
  inner loop i: 1
    DB 호출 없는 Mono: 1
  inner loop i: 2
    DB 호출 없는 Mono: 2
  inner loop i: 3
    DB 호출 없는 Mono: 3
  inner loop i: 4
    DB 호출 없는 Mono: 4
Second doOnNext
outer loop index: 1
  inner loop i: 1
    DB 호출 없는 Mono: 1
  inner loop i: 2
    DB 호출 없는 Mono: 2
  inner loop i: 3
    DB 호출 없는 Mono: 3
  inner loop i: 4
    DB 호출 없는 Mono: 4
Second doOnNext
doOnComplete i: 1
doOnComplete i: 2
doOnComplete i: 3
doOnComplete i: 4
doOnComplete i: 5
    DB 호출 있는 Mono: moid117 - 문어
    DB 호출 있는 Mono: moid117 - 문어
    DB 호출 있는 Mono: moid117 - 문어
    DB 호출 있는 Mono: moid117 - 문어
    DB 호출 있는 Mono: moid117 - 문어
    DB 호출 있는 Mono: moid117 - 문어
    DB 호출 있는 Mono: moid117 - 문어

‘DB 호출 없는 Mono’는 예상대로 로그에 찍혀 있지만, 함께 찍힐 것이라고 기대했던 ‘DB 호출 있는 Mono’는 가장 마지막에 찍혀 있다.

코드를 보면 DB 조회를 모두 완료한 후에 이름처럼 doOnComplete(...)과 then(...)이 실행될 것 같지만, 실제로는 DB 조회 결과 후 리스트에 추가하는 부분이 가장 나중에 실행된 것이다.
심지어 메서드가 반환된 다음에 실행되기 때문에 비어있는 리스트가 반환됐다.

그뿐만이 아니다. Flux.range(0, 2).doOnNext(...)는 예상대로 0, 1만 반복하지만, Flux.range(1, 4).subscribe(...)는 예상과 달리 1, 2, 3에서 끝나지 않고 4까지 반복된다. range()의 두 번째 인자가 inclusive 인지 exclusive 인지도 그때그때 달라요 처럼 보인다. - 추가 Flux.range()는 Flux.range(start, count)라서 이렇게 동작한다고 양봉수 님께서 알려주셨다. 덕분에 왜 저렇게 동작하는지는 알게 됐지만 IntStream.range(startInclusive, endExclusive)를 비롯해서 비슷한 API들의 2번째 인자가 대부분 endExclusive라는 걸 감안하면 Flux.range()는 잘 알려진 관습과 달라서 직관성이 떨어지며 불필요한 학습을 요구하고 있다.

이처럼 직관과 다른 부분이 Reactor(RxJava도 마찬가지) 사용을 힘들게 만든다. 물론 앞서 말했 듯 학습을 통해 제대로 된 사용법을 익히면 |로 구분된 문자열을 반환하도록 수정할 수 있겠지만 그 학습 비용이 만만치 않다. 게다가 대안이 있다면 더더욱 그 비용은 낭비다.

Coroutine

똑같은 일을 하는 코루틴 코드를 보자. suspend와 awaitSingle()을 제외하면 늘 봐오던 코드와 다를 게 없이 편안하다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

suspend fun asyncAwaitRange(): String {
    val strList: MutableList<String> = mutableListOf()

    for (outer in 0..1) {
        println("outer loop index: $outer")

        for (inner in 1..4) {
            println("  inner loop i: $inner")
            println("    DB 호출 없는 Mono: $inner")
            val member = memberRepository.findById("moid117").awaitSingle()
            println("    DB 호출 있는 Mono: ${member.oid} - ${member.name}")
            strList.add("${member.oid} - ${member.name}")
        }

        println("Second doOnNext")
    }

    for (i in 1..5) {
        println("doOnComplete i: $i")
    }

    return "[${strList.joinToString(" | ")}]"
}

아래와 같은 컨트롤러를 붙여서,

1
2
3
4

@GetMapping("/async-await-range")
suspend fun asyncAwaitRange(): ResponseEntity<String> {
    return ResponseEntity.ok(memberCommandService.asyncAwaitRange())
}

아래와 같이 호출해보면 의도했던 것처럼 문자열 목록이 나온다.

1
2
3
4
5
6

~ 🦑🍕🍺 ❯ http GET :8080/api/members/async-await-range
HTTP/1.1 200 OK
Content-Length: 151
Content-Type: text/plain;charset=UTF-8

[moid117 - 문어 | moid117 - 문어 | moid117 - 문어 | moid117 - 문어 | moid117 - 문어 | moid117 - 문어 | moid117 - 문어 | moid117 - 문어]

로그도 직관에서 한 치의 벗어남이 없다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

outer loop index: 0
  inner loop i: 1
    DB 호출 없는 Mono: 1
    DB 호출 있는 Mono: moid117 - 문어
  inner loop i: 2
    DB 호출 없는 Mono: 2
    DB 호출 있는 Mono: moid117 - 문어
  inner loop i: 3
    DB 호출 없는 Mono: 3
    DB 호출 있는 Mono: moid117 - 문어
  inner loop i: 4
    DB 호출 없는 Mono: 4
    DB 호출 있는 Mono: moid117 - 문어
Second doOnNext
outer loop index: 1
  inner loop i: 1
    DB 호출 없는 Mono: 1
    DB 호출 있는 Mono: moid117 - 문어
  inner loop i: 2
    DB 호출 없는 Mono: 2
    DB 호출 있는 Mono: moid117 - 문어
  inner loop i: 3
    DB 호출 없는 Mono: 3
    DB 호출 있는 Mono: moid117 - 문어
  inner loop i: 4
    DB 호출 없는 Mono: 4
    DB 호출 있는 Mono: moid117 - 문어
Second doOnNext
doOnComplete i: 1
doOnComplete i: 2
doOnComplete i: 3
doOnComplete i: 4
doOnComplete i: 5

마무리

직관에서 많이 벗어나는 Reactive Streams는 써야만 할 때만 쓰자.

써야만 할 때가 언제일까?

https://www.reactive-streams.org/ 의 첫 문장에 다음과 같이 나와있다.

Reactive Streams is an initiative to provide a standard for asynchronous stream processing with non-blocking back pressure.

대충 우림말로 ‘논블로킹 배압을 사용해서 비동기 스트림을 처리하는 표준을 제공한다’ 정도로 보이는데 길진 않지만 금방 이해가 되지는 않는다.

정확하게 같은 의미는 아닐지라도 앞에서 살펴본 사례에서 이어지는 흐름대로 조금 구체화해서 재해석해보면,

선형적인 처리 흐름인데 그 흐름에 들어있는 단위 과정 중간중간에 자원 낭비를 유발하는 blocking 구간이 있어서 이를 최적화 해야하고 배압(back pressure) 적용이 필요할 때

라고 할 수 있을 것 같다.

물론 선형적인 흐름이 아니라도 Mono.zip() 등 여러가지 (너무) 다양한 리액티브 연산자를 동원하면서 의도대로 실행되게 작성할 수는 있겠지만, 많은 학습 비용과 삽질을 각오해야 한다. 할 수 있다고 해서 무조건 하는 거랑 대안을 검토해보고 해야만 할 때 하는 거랑은 큰 차이가 있다.

이 글에서는 언급하지 않았지만 Reactive Streams에서 강조하는 또 한 가지 요소는 배압(back pressure)이다. 배압은 데이터를 발생시키는 Publisher(Producer)에게 Subscriber(Consumer)가 처리할 수 있는 만큼만 요청할 수 있게 해주는 메커니즘을 의미한다.

일반적인 웹 서비스에서 보면 웹 클라이언트가 서버에 요청을 보내면(Publish), API 서버는 이를 받아서 처리(Consume)하고 응답을 클라이언트에게 반환한다. 이런 흐름에서 API 서버가 처리할 수 있는 만큼만 보내라는 요청을 개별 웹 클라이언트에게 보내는 일은 거의 없다.

결국 위 코드 사례처럼 선형적인 처리가 아니라 여러 중첩된 처리 과정이 난무할 때, 그리고 배압 메커니즘이 필요 없을 때, 그러니까 사실 상 대부분의 일반적인 웹 서비스의 경우에는 코루틴을 사용하는 것이 편리하다고 본다.

그리고 머지 않아(한 2년 이내?) Virtual Thread가 나온다는 것을 감안하면 Reactive Streams를 사용한 코드는 많은 경우 부담스러운 레거시로 남을 가능성이 크다.

이미 Reactive Streams로 작성되어 있어 피할 수 없거나, 또는 비동기 스트림 처리와 배압이 정말로 필요한 상황이라서 써야만 한다면, https://homoefficio.github.io/2021/04/14/Reactive-Streams-with-Sequence-Diagram/ 여기에서 시작해보자.

1 req == 1 therad인 서블릿 방식의 한계를 뛰어넘기 위해 Spring에서 WebFlux를 내놨다.
Spring WebFlux는 내부적으로 Reactor를 사용하는데, Reactor는 Reactive Streams 구현체다.
Reactive Streams는 홈페이지에 다음과 같이 간단 명료하게 정의돼 있다.

Reactive Streams is an initiative to provide a standard for asynchronous stream processing with non-blocking back pressure.

리액티브 스트림은 비동기 스트림 처리 표준을 제공하는 킹왕짱 계획이야. 논블로킹 백프레셔도 있다능!

솔직히 뭔 소린지 모르겠다. 이거 안다고 퇴근 시간이 앞당겨질 것 같지는 않은데 몰라도 너무 모르니 한 번 알아보려 한다.

이후 나오는 내용은 JDK 9 Flow 예제와 Reactor, Reactive MongoDB에서 구현한 여러 Reactive Streams 구현 내용을 기준으로 작성했으며, Hot Publisher는 배제하고 Cold Publisher만 다룬다.

등장 인물

Reactive Streams는 Publisher, Subscriber, Subscription, Processor 달랑 네 개의 인터페이스로 구성돼있다.
이 중에서 Processor는 자기만의 고유한 메서드는 없고 단순히 Publisher, Subscriber 두 가지 인터페이스를 상속받는다. Processor 없이 나머지 3개만으로도 Reactive Streams 협력을 구성할 수 있으므로 Processor는 등장 인물에서 제외한다.

얘네들이 각각 무슨 동작을 가지고 있고 어떻게 협력해서 클라이언트에게 데이터를 반환하는지 한 눈에 알아보자.

시퀀스 다이어그램

검색해보면 몇 가지 나오기는 하는데 내가 이해할 수 있을 만큼 마음에 드는 게 없어서 새로 그렸다(개고생 ㅠㅜ).

Publisher, Subscriber, Subscription 인터페이스가 가지고 있는 모든 메서드가 표시돼 있다. publisher 의 map, flatMap, zip, ...는 Reactive Streams 명세에 있는 메서드가 아니라 구현체인 Reactor의 Flux에 있는 메서드인데 설명의 편의를 위해 추가했다.

Reactive Streams를 사용해서 비동기로 데이터를 조회하는 시나리오를 기준으로 작성했으며,
실선 화살표는 메서드 호출, 화살표 위는 메서드 이름, 화살표 아래에 괄호로 표시된 건 메서드 인자이며, 점선 화살표는 반환이고 화살표 아래 괄호 없이 표시된 건 반환값이다.

데이터 핸들러 로직 정의 및 Subscriber 생성

데이터를 요청하는 Client는 데이터를 받아서 어떻게 처리할지, 받는 과정에서 오류를 전달 받으면 어떻게 처리할지, 데이터를 모두 받은 후에 어떤 일을 할지 정해야 할 책임을 가지고 있다. 그런 책임을 각각 nextConsumer, errorConsumer, completeRunnable로 정의해서 이를 바탕으로 subscriber를 생성한다.

설명의 편의를 위해 subscribe 생성을 가장 먼저 표시했는데 그림을 보면 알 수 있겠지만 반드시 가장 먼저 수행할 필요는 없다. publisher.subscribe(subscriber)를 호출하기 직전에 subscriber를 생성해도 된다.

Data Provider에 데이터 요청 및 Publisher 생성

클라이언트는 DataProvider에게 데이터를 요청한다. DataProvider는 특정 클래스 이름은 아니고 클라이언트로부터 호출을 받으면 데이터 저장소와 연동해서 실제 데이터를 반환하는 책임이 있는 객체를 의미한다고 보면 된다. 예를 들면 ReactiveMongoOperations(ReactiveMongoTemplate)이나 ReactiveMongoRepository라고 생각하면 된다. 이 DataProvider는 나중에 데이터를 제공할 수 있도록 콜백을 생성하고 이를 publisher를 생성하면서 주입해준다. 이 부분 자세한 과정은 맨 아래에서 구경할 수 있다. DataProvider는 생성한 publisher를 클라이언트에게 반환한다.

구독

클라이언트는 DataProvider로부터 publisher를 반환 받은 후에는, 나중에 publisher가 발행할 데이터를 받아서 비즈니스 요구에 맞게 가공하는 로직을 추가한다. map, flatMap, zip 등 여러 리액티브 연산자가 이 때 사용된다.

클라이언트는 데이터 가공 로직 추가를 마친 후에publisher.subscribe(subscriber)를 호출한다. 리액티브 스트림에서 절대 잊어서는 안 될 가장 중요한 특징 중 하나가 바로 구독하기 전에는 아무 일도 일어나지 않는다는 점이다. 즉 앞에서 아무리 nextConsumer, errorConsumer, completeRunnable를 모두 정의하고, DataProvider를 호출해서 데이터를 가져오고 가공하는 로직을 구현해뒀다 하더라도, publisher.subscribe(subscriber)를 호출하지 않으면 앞서 만든 모든 것들은 전혀 실행되지 않는다. 더 정확하게 말하면 데이터를 가져오는 로직은 아직 콜백에 담겨 있을 뿐이고, 구독하기 전에는 콜백이 실행되지 않는다.(엄밀하게는 Hot Publisher인 경우 구독과 상관 없이 데이터를 뿜어내지만 앞서 얘기했듯이 Hot은 너무 뜨거워서 생략)

Subscription 생성

publisher.subscribe(subscriber)가 호출되면 publisher는 인자로 전달받은 subscriber와 자신이 생성될 때 주입받은 dataCallback를 바탕으로 subscription을 생성하고, subscriber.onSubscribe(subscription)를 호출해서 subscription을 subscriber에게 전달해준다.

Subscription에 데이터 요청

subscriber는 onSubscribe(subscription)을 통해 subscription을 전달받으면 subscription.request(numOfData)를 호출해서 데이터를 요청한다. 자신이 소화할 수 있을 만큼의 데이터만 요청할 수 있으므로 back pressure 개념이 이 지점에서 발동한다. 그리고 실제 데이터 접근도 이 시점에서 이루어진다.

실제 데이터 접근 및 onNext/onError/onComplete 호출

subscription은 자신이 생성될 때 주입 받은 콜백을 호출해서 numOfData만큼만 데이터를 가져오고 subscriber.onNext(data)를 반복 호출해서 subscriber에게 데이터를 전달한다. 이 과정에서 오류가 발생하면 subscriber.onError(throwable)로 오류를 subscriber에게 전달하고, 데이터 전달이 정상적으로 완료되면 subscriber.onComplete()를 호출하며 협력이 끝난다.

비동기는 대체 어디에?

리액티브 스트림이 백프레셔와 함께 비동기 스트림 처리 표준을 제공하는 킹왕짱 계획이라고 했는데, 이 협력 구조 상에서 비동기 처리는 어디에 있는 걸까?

사실 리액티브 스트림이 비동기 스트림 처리 표준 제공 어쩌구라고는 하지만 4가지 인터페이스를 보면 비동기 관련 내용은 전혀 없다. 다시 말해 비동기 처리 없이 동기 처리만 사용하더라도 스트림을 리액티브 방식으로 처리하는 것이 가능하다. 결국 리액티브 스트림은 비동기 처리 표준을 지향하긴 하지만 그렇다고 비동기를 강제하는 것도 아니다. 따라서 비동기 처리는 실질적으로는 구현에 달려 있다.

Reactor의 비동기 처리 관련 규약은 reactor.core.scheduler.Scheduler 인터페이스에 담겨 있다. reactor.core.publisher 패키지에서 Scheduler가 사용되는 곳을 검색하면 어떻게 비동기 처리를 하는지 대략 감을 잡을 수 있을 것이다.

또 다른 예로 JDK 9 예제에 보면 Subscription이 Subscriber의 onNext()를 호출할 때 Executor를 이용해서 비동기로 호출하고 있다.

마무리

Reactive Streams는 논블로킹, 백프레셔를 포함해서 스트림을 비동기 방식으로 처리할 수 있는 표준 API다.

Publisher, Subscriber, Subscription 이 서로 협력하면서 스트림을 리액티브 방식으로 처리한다.
자세한 협력 구조는 글보다는 시퀀스 다이어그램을 참고하면 더 쉽게 이해할 수 있다.

번외편

사실 여기까지만 알면 리액티브 스트림의 협력 구조를 이해하는 데는 충분하다고 생각한다.
물론 그렇다고 리액티브 스트림을 사용하는 데 충분하다는 얘기는 아니다. 실무적으로는 map, flatMap, concatMap, zip 등등 엄~~청나게 많은 리액티브 연산자들 예쁘디 예쁜 마블 다이어그램 보면서 다 공부해야 된다능~~ ㄷㄷㄷ

암튼 원래는 이 정도만 알고 넘어가려고 했는데.. 이 정도 알아보고 나니 또 다른 궁금증이 파생되어..
아래에 이어지는 건 굳이 안 봐도 된다. 그냥 개인적인 주절거림일 뿐이다.

이런 이름 적절한가?

Publisher를 보자. 퍼블리셔라면 뭔가 데이터를 발행하고 뿜어내야 할 것 같은데 정작 가진 메서드 이름이나 파라미터는 영 다르다. 달랑 하나 있는 메서드는 다음과 같다.

Publisher.subscribe(Subscriber)

발행자가 구독한다 구독자를. 읭? 발행자가 구독자를 구독한다고? 뭥미?

일단 구독은 구독자의 행위인데 구독자가 아니라 엉뚱하게 발행자가 주어로 나와있다. 사실 이렇게 행위의 주어와 메서드가 소속된 객체가 다르게 돼 있는 API는 많다. 모든 getter 메서드는 get이라는 행위의 주어와 getXXX라는 메서드가 소속된 객체가 다르다. 멀리 갈 것 없이 위 시퀀스 다이어그램에 나오는 subscription.request(numOfData)도 마찬가지다. request 행위의 주어는 subscriber지만 request() 메서드는 subscription에 소속돼 있다. 그래도 이 코드를 읽는 데는 전혀 불편함이 없다. 결국 행위의 주어와 메서드가 소속된 객체가 다르다는 것만으로는 불편함을 느끼지 않는다.

하지만 Publisher.subscribe(Subscriber)는 단순히 행위의 주어와 메서드가 소속된 객체가 불일치하는 데 그치지 않고, 발행자/구독자라는 대칭 관계에 있는 애들이 구독한다(subscribe)라는 행위(동사)를 중심으로 주어, 목적어가 뒤바뀌어 있다. 그래서 직관적으로 자연스럽게 협력 구조를 이해하는 데 큰 걸림돌이 된다.

단순히 문장 해석하듯 자연스럽게 이해되지 않는 것뿐만 아니라 불편한 이유는 몇 가지 더 있다.

먼저 이 subscribe라는 이름은 실제 동작과도 부합하지 않는다.

JDK API 문서에 나오는 Publisher.subscribe(Subscriber)에 대한 설명은 다음과 같은 간단하고 명료한 문장으로 시작한다.

Adds the given Subscriber if possible.

즉 실제 하는 동작은 Subscriber를 add 하는 건데 subscribe 라는 엉뚱한 이름이 사용되고 있다.

게다가 Reactive는 Push 기반 데이터 처리 패턴이라고 할 수 있는데, 데이터 처리 실행을 유발하는 핵심 동작을 구독자 입장의 동작인 subscribe라고 이름지어버리니까 Publisher가 Push하는 패턴이 아니라 Subscriber가 Pull하는 패턴처럼 느껴진다. 어쩌면 Push를 퇴색시키는 이 부분이 Reactive를 이해하고 사용하는 데 있어 가장 나쁜 점일 수도 있겠다. 이쯤되면 불편함을 넘어 해로움이라고 할 수도 있겠다.

불편했던 또 한 가지 이유는 onSubscribe, onNext, onError, onComplete 메서드의 컨텍스트인 Subscriber는 주어의 역할을 하는데, Subscriber와 대칭 관계라고 할 수 있는 Publisher는 subscribe 메서드의 주어도 아니고 목적어도 아니니 대칭성이 깨져서 직관성이 떨어지기 때문이다.

암튼 내가 보기엔 불편함을 넘어 해로워보이기까지 하는 subscribe(Subscriber)가 사용된 이유는 아마도 답정너처럼 Reactive Streams 협력 구조에서 subscribe라는 용어를 중심으로 내세우고 싶었던 것이 아니었을까 짐작해본다.

예전에도 이 부분이 마음에 안 들어서 https://www.facebook.com/hanmomhanda/posts/10214512128821140 여기에 끄적여둔 게 있다. 댓글들을 보면 나를 제외한 나머지 모두는 Publisher.subscribe(Subscriber) 이게 어색하지도 해석하는 데 불편하지도 않다는 반응이라서 신기하기도 했다.

Publisher.subscribe(Subscriber)가 불편하다면, Publisher/Subscriber와 비슷하게 Producer/Consumer 관계가 사용되는 Stream.forEach(Consumer)도 불편해야 하는 거 아니냐는 날카로운 의견도 있었는데, 일단 Stream이 역할은 Producer 이긴 하지만 API 상 공식 이름은 어디까지나 Stream이다. Stream.forEach(Consumer)는 스트림 안에서 흐르는 각 원소에 대해 Consumer가 소비한다 외에 다른 해석을 떠올리기 어려울 정도로 직관적이므로 Publisher.subscribe(Subscriber)와는 많이 다르다고 생각한다.

cancel은?

실제 구독을 중단하는 일을 수행하는 책임은 현재 Subscription에 있다. 그리고 Subscription.cancel()를 누가 호출하는지 구현체를 살펴보니 본 것 중에서는 모두 Subscriber가 호출하고 있다. 하지만 현재 설계 상으로는 Subscriber에 cancel 관련 공개 메서드가 없으므로, cancel 여부를 결정하는 클라이언트가 cancel을 유발할 수 있는 방법이 없다.

그럼 도대체 cancel이 어떻게 실현될 수 있는 걸까? 구현체인 Reactor를 살펴보니, Reactive Streams 에서는 Publisher.subscribe()는 리턴 값이 없지만 리액터의 Publisher 구현체인 Flux에는 Disposable을 반환하는 subscribe() 메서드가 있다. 이 Disposable에 dispose() 메서드가 있어서 이를 통해 Subscription.cancel()을 호출할 수 있다. 그런데 Disposable도 Reactive Streams에는 없고 구현체인 Reactor에서 만들어 사용하는 인터페이스다.

요는 Reactive Streams만으로는 cancel 여부를 결정하는 클라이언트가 cancel 할 수 있는 방법이 없다.

개선된 시퀀스 다이어그램

궁시렁대기만 할 게 아니라 개선도 한 번 생각해보자.

subscribe라는 용어를 중심에 두고 싶었다면 협력 구조도 좀 바꿔서 Subscriber.subscribe(Publisher)로 했으면 어땠을까? 이렇게 하면 클라이언트가 Subscriber와만 직접 통신하게 되고, 너무 구독만 강조돼서 Push 기반이라는 Reactive의 특징이 퇴색되는 안 좋은 결과로 이어진다. 게다가 코드도 아래와 같이 못난이가 돼버린다.

1
2
3
4
5
6

subscriber.subscribe(
    Flux.just("a", "b", "c")
        .filter(...)
        .map(...)
        ...
);

결국 컨텍스트인 Publisher를 Subscriber로 바꾸는 건 좋은 대안이 아닌 걸로 보인다. 그렇다면 Publisher는 그대로 두고 subscribe(Subscriber) 대신 실제 동작에 맞게 add(Subscriber)로 하거나, add가 너무 일반적인 동사라 의도를 표현하는 데 부족하다면 addSubscriber(Subscriber)로 했다면 훨씬 이해하기 편했을 것 같다. 그런데 이렇게 하면 ‘구독하지 않으면 아무 일도 일어나지 않는다’라는 문장과 어긋나는 면이 있고, 여전히 계속 구독만 강조되는 것처럼 보인다.

‘추가’라는 표면적인 동작과는 맞지 않지만 결국에는 구독자에게 발행한다는 의미에 무게를 둬서 publishTo(Subscriber)로 하면 어떨까? 일단 이렇게 하면 ‘발행자가 구독자에게 발행한다’라고 아주 직관적으로 이해할 수 있게 된다. 게다가 Push 모델이라는 점도 훨씬 극명하게 드러난다. ‘구독하지 않으면 아무 일도 일어나지 않는다’ 대신에 ‘발행하지 않으면 아무 일도 일어나지 않는다’로 바꿔도 전혀 어색하지 않다.

이름 하나 바꿨을뿐인데 모든 게 좋아진 것 같(은 환각이 든)다. cancel도 Publisher에게 추가하는 것이 좋겠다.

이를 반영해서 개선한 시퀀스 다이어그램은 다음과 같다. 딸기색 둥근 네모 부분만 달라졌다.

비동기 처리 관점에서 리액티브 스트림의 가치

리액티브 스트림을 활용한 프로그래밍은 여러모로 진입 장벽이 높다. 그런데 그걸 꼭 넘어서 사용해야할 정도로 가치가 있을까?

비동기 처리라면 C#, JavaScript, Rust 등에는 async/await, Kotlin에는 coroutine 처럼 더 진입 장벽이 낮은 API가 제공되고 있다. 그리고 자바에도 정확히 언제가 될지는 모르지만 Fiber(Project Loom)가 도입될 예정이므로 비동기 처리라는 관점에서 리액티브 스트림이나 ReactiveX가 앞서 예를 든 더 간편한 API들과 견주어 경쟁력을 유지할 수 있을지 솔직히 의문이다. 한 예로 Kotlin Coroutine과 Reactive Streams 코드를 비교한 자료(https://github.com/HomoEfficio/dev-tips/blob/master/Kotlin-Coroutine-vs-Reactive-Streams(Reactor).md) 를 보면 이런 의문을 가질 법하다는 사실을 실감나게 느낄 수 있을 것이다.

따라서 비동기 처리 관점에서 리액티브 스트림을 아주 깊게 이해해야만 할 것 같지는 않다. 그저 back pressure를 적용할 수 있어야 하고, onNext, onError, onComplete 와 같이 이벤트 핸들링 방식으로 처리하는 API를 제공하려다보니 이런 설계가 나왔겠지 정도로 털고 가자(아 훈훈해..).

물론 꼭 비동기 처리를 필요로 하지 않는 Push 기반의 데이터 처리 패턴으로서의 존재 가치는 여전히 유효할 것이다.

콜백이 어디에 어떻게 감춰져 있는지 구경하기

그럼 Reactor와 Reactive MongoDB가 추상화해서 감춰든 부분을 한 번 구경해보자. 위에서 아래로 호출이 이어진다고 보고 흐름을 구경하면 된다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134

// ReactiveMongoTemplate

    protected <T> Mono<T> doFindOne(String collectionName, Document query, @Nullable Document fields,
            Class<T> entityClass, FindPublisherPreparer preparer) {

        MongoPersistentEntity<?> entity = mappingContext.getPersistentEntity(entityClass);

        QueryContext queryContext = queryOperations
                .createQueryContext(new BasicQuery(query, fields != null ? fields : new Document()));
        Document mappedFields = queryContext.getMappedFields(entity, entityClass, projectionFactory);
        Document mappedQuery = queryContext.getMappedQuery(entity);

        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug(String.format("findOne using query: %s fields: %s for class: %s in collection: %s",
                    serializeToJsonSafely(query), mappedFields, entityClass, collectionName));
        }

        // 여기!!
        return executeFindOneInternal(new FindOneCallback(mappedQuery, mappedFields, preparer),
                new ReadDocumentCallback<>(this.mongoConverter, entityClass, collectionName), collectionName);
    }
    

    private <T> Mono<T> executeFindOneInternal(ReactiveCollectionCallback<Document> collectionCallback,
            DocumentCallback<T> objectCallback, String collectionName) {

        // 여기!!
        return createMono(collectionName,
                collection -> Mono.from(collectionCallback.doInCollection(collection)).flatMap(objectCallback::doWith));
    }


    public <T> Mono<T> createMono(String collectionName, ReactiveCollectionCallback<T> callback) {

        Assert.hasText(collectionName, "Collection name must not be null or empty!");
        Assert.notNull(callback, "ReactiveCollectionCallback must not be null!");

        Mono<MongoCollection<Document>> collectionPublisher = doGetDatabase()
                .map(database -> getAndPrepareCollection(database, collectionName));

                                                                   // 여기!!
        return collectionPublisher.flatMap(collection -> Mono.from(callback.doInCollection(collection)))
                .onErrorMap(translateException());
    }



    private static class FindOneCallback implements ReactiveCollectionCallback<Document> {

        private final Document query;
        private final Optional<Document> fields;
        private final FindPublisherPreparer preparer;

        FindOneCallback(Document query, @Nullable Document fields, FindPublisherPreparer preparer) {
            this.query = query;
            this.fields = Optional.ofNullable(fields);
            this.preparer = preparer;
        }

        @Override
        public Publisher<Document> doInCollection(MongoCollection<Document> collection)
                throws MongoException, DataAccessException {

            if (LOGGER.isDebugEnabled()) {

                LOGGER.debug("findOne using query: {} fields: {} in db.collection: {}", serializeToJsonSafely(query),
                        serializeToJsonSafely(fields.orElseGet(Document::new)), collection.getNamespace().getFullName());
            }

                                                        // 여기!!
            FindPublisher<Document> publisher = preparer.initiateFind(collection, col -> col.find(query, Document.class));

            if (fields.isPresent()) {
                publisher = publisher.projection(fields.get());
            }

            return publisher.limit(1).first();
        }
    }



// FindPublisherPreparer
    default FindPublisher<Document> initiateFind(MongoCollection<Document> collection,
            Function<MongoCollection<Document>, FindPublisher<Document>> find) {

        Assert.notNull(collection, "Collection must not be null!");
        Assert.notNull(find, "Find function must not be null!");

        if (hasReadPreference()) {
            collection = collection.withReadPreference(getReadPreference());
        }
               // 여기!!
        return prepare(find.apply(collection));
    }


// ReactiveMongoTemplate
        public FindPublisher<Document> prepare(FindPublisher<Document> findPublisher) {

            FindPublisher<Document> findPublisherToUse = operations.forType(type) //
                    .getCollation(query) //
                    .map(Collation::toMongoCollation) //
                    .map(findPublisher::collation) //
                    .orElse(findPublisher);

            // findPublisherToUse 에 limit, skip, hint 등 적용한 후에 반환
            // ...

            return findPublisherToUse;
        }


// FindPublisehrImpl
final class FindPublisherImpl<TResult> implements FindPublisher<TResult> {

    private final AsyncFindIterable<TResult> wrapped;

    FindPublisherImpl(final AsyncFindIterable<TResult> wrapped) {
        this.wrapped = notNull("wrapped", wrapped);
    }

    @Override
    public Publisher<TResult> first() {
               // 여기!!
        return Publishers.publish(wrapped::first);
    }


//Publishers
public static <TResult> Publisher<TResult> publish(final Block<SingleResultCallback<TResult>> operation) {
           // 여기!!
    return subscriber -> new SingleResultCallbackSubscription<>(operation, subscriber);
}

DZone에서 본 글인데 동시성 처리 관련 여러 방식을 비교하면서, 프로젝트 룸(Loom) 코드도 구경할 수 있어서 원글 작성자의 허락을 받고 우리말로 옮겨본다.

원문: https://dzone.com/articles/java-concurrency-evolution

자바는 초창기부터 스레드를 사용해서 동시성 프로그래밍을 할 수 있었다. 자바 1.1 버전까지는 JVM에서 그린 스레드(가상 스레드)를 지원했지만 그 이후 버전에서는 폐기되고 OS 네이티브 스레드를 사용하게 됐다. 하지만 오랫동안 관 속에 묻혀있던 가상 스레드를 다시 부활시키는 프로젝트 룸(Loom)이 수면 위로 부상하면서, 폐기됐던 가상 스레드가 다시 부활하여 주류에 올라설 수 있게 됐다.

이 글의 목적은 자바의 스레드/동시성 처리 진화 과정에 있었던 주요 마일스톤을 살펴보는 것이다. 스레드와 동시성을 다루는 자료는 차고 넘쳐나므로 이 글의 목적에서 벗어나는 다음 내용은 여기에서는 다루지 않는다.

• 에러 처리: 거의 다루지 않으며 코드 가독성을 높이기 위해 Lombok의 @SneakyThrows만 사용한다.
• 동시성 라이브러리나 프레임워크: JVM에는 Quasar, Akka, Guava, EA Async 등 동시성을 다루는 라이브러리와 프레임워크가 아주 많다.
• 복잡한 종료 조건: 주어진 태스크(task)를 수행하는데 필요한 스레드를 시작하고 나서, 스레드 종료까지 얼마나 기다려야 하는지 항상 명확하지는 않다.
• 스레드 간 동기화: 이 얘기를 다뤘다간 글을 끝맺을 수 없을 것만 같다.

그럼 이 글에서 다루는 건 뭘까? 좋은 질문이다. 재미있는 건 위에서 다 빼버렸는데 무슨 얘기를 하려는 걸까?

똑같은 일을 하는 예제 하나를 옛날 방식부터 시작해서 자바 동시성 진화 과정을 따라 더 새롭고 색다른 방법으로 구현하면서 그 실행 ‘동작’(behavior)을 비교해보려고 한다. 여기서 다루는 방식이 전부는 아니며 다른 방식도 있다. 자바 언어에서 제공하는 API만으로 구현할 수 있는 방법만을 모아서 살펴보려고 했는데, 주류라고 할 수 있는 리액티브 방식을 외면할 수 없어서 같이 다루기로 한다.

자바 스레드

본격적으로 시작하기 전에 자바의 스레드에 대해 몇 가지만 짚고 넘어가자.

• JVM 스레드와 OS 스레드는 1:1로 매핑된다. JVM 스레드는 OS 스레드를 얇게 덧씌워 만든 거라고 볼 수 있다.
• OS는 아주 범용적인(그래서 느린) 스케줄링을 사용한다. OS는 JVM 내부에 대해 아무 것도 알지 못한다.
• 스레드를 만들고 이 스레드 저 스레드를 오가는 작업은 커널을 거쳐야 하므로 비용이 많이 든다(느리다).
• OS의 continuation 구현체는 자바 콜 스택(call stack) 뿐만 아니라 네이티브 콜 스택도 포함하며, 자원을 많이 사용한다.
• OS 스레드 갯수는 CPU 코어 숫자에 의해 제약받는다.
• 스레드에 사용되는 스택 메모리는 OS에 의해 힙 외의 영역에 마련된다.

태스크

예제에서 수행할 태스크(task)는 주로 동시에 호출되어 실행된다. 사용자별로 다음과 같은 흐름으로 요청을 처리하는 웹 서버를 떠올려보자.

• 서비스 A가 호출되면 완료될 때까지 1000ms가 필요하다.
• 서비스 B가 호출되면 완료될 때까지 500ms가 필요하다.
• 서비스 A, B의 결과는 파일, DB, S3 등 저장 방식별로 Z회 저장되며, 한 번 저장하는데 300ms가 필요하다. 현실에서는 저장 방식 별로 소요 시간이 다르겠지만 계산의 편의를 위해 같다고 가정한다.

동시 실행이 전혀 없는 No Concurrency 방식에서는 요청 갯수 * (서비스A 처리 시간 + 서비스B 처리 시간 + 저장 횟수 * 저장 시간)만큼의 시간이 필요하다.

반면에 모든 요청이 동시에 실행되는 이상적인 Full Concurrency 방식에서는 Max(서비스A 처리 시간, 서비스B 처리 시간) + 저장 시간 만큼, 그러니까 1,300ms가 필요하다.

서비스 처리나 저장에 필요한 시간은 단순하게 Thread.sleep()으로 구현했다. 현실성은 물론 떨어지지만 여러 방식을 비교하고 이해하는 데는 충분하다.

전체 코드는 여기에서 확인할 수 있다. 요청 갯수(N)와 저장 횟수(Z)를 마음대로 바꿔가면서 어떤 결과가 나오는지 살펴보자. 늘 그렇듯이 모든 문제는 여러 가지 방식으로 풀 수 있으며, 깃헙에 있는 코드는 그 중 하나일 뿐이다. 가독성을 위해 최적화를 희생한 코드도 있음을 미리 밝혀둔다. 이제 예제 실행 결과를 살펴보면서 재미나는 자바 동시성 진화 과정을 따라가보자.

동시성 미사용

가장 단순한 방식이며, 아주 친숙한 코드다. 이런 코드 작성을 도와주는 도구도 많고, 쉬워서 직관적으로 금방 이해하고 디버깅 할 수 있다.

하지만 자원을 효율적으로 사용하지 못해서 실행 성능은 떨어진다는 것을 쉽게 알 수 있다. 사용된 유일한 JVM 스레드는 하나의 OS 스레드를 사용하며, 하나의 OS 스레드는 하나의 CPU 코어를 사용하므로 나머지 코어는 모두 놀게 된다. 요청 갯수와 저장 횟수가 많아지면 실행 소요 시간도 계속 늘어난다.

Code

코드도 아주 직관적이다. 짧고 핵심만 들어있다.

1
2
3
4
5
6
7
8
9
10

public void shouldBeNotConcurrent() {

    for (int user = 1; user <= USERS; user++) {
        String serviceA = serviceA(user);
        String serviceB = serviceB(user);
        for (int i = 1; i <= PERSISTENCE_FORK_FACTOR; i++) {
            persistence(i, serviceA, serviceB);
        }
    }
}

helper 메서드 내용은 가독성을 위해 생략했는데 원한다면 여기에서 확인할 수 있다.

실행 결과

실행 결과는 앞의 ‘Task’ 단원에 나온 그림 아래 쪽 표의 No Concurrency와 같다.

네이티브 멀티 스레딩

멀티 스레딩에는 일반적으로 다음과 같은 난관이 있다.

• CPU 코어나 메모리 같은 자원의 효율적 이용
• 세밀한 스레드 갯수 조절이나 스레드 관리
• 제어 흐름과 컨텍스트 유실
• 실행 동기화
• 디버깅과 테스트

이 중 세 번째 항목인 ‘제어 흐름과 컨텍스트 유실’이 발생하는 이유는, 스택 트레이스가 요청 처리 전체를 아우르는 트랜잭션이 아니라, 그 중 일부만을 처리하는 스레드에 바운드 되므로, 현재 스레드에 할당된 일부 단계에 대한 정보만 스택 트레이스로 확인할 수 있기 때문이다. 그래서 디버깅이나 프로파일링이 어려워질 수 있다.

코드

무엇보다도 일단 코드 량이 확연히 늘어난 것을 볼 수 있다. Runnable을 구현해야 하고, 수동으로 스레드를 생성하고 제어하며, 동기화도 필요하다. 로직이 여기저기 흩어져 있어 따라가기가 어렵다.

반면에 실행 성능은 꽤 좋다. 스레드 생성과 컨텍스트 스위칭 비용이 있으므로 이상적인 수치인 1,300ms에는 못 미치지만 앞서 살펴본 ‘동시성 미사용’ 방식에 비하면 훨씬 좋다.

아래는 가독성을 위해 일부만 가져왔으며 네이티브 멀티 스레딩 방식 전체 코드는 여기에서 확인할 수 있다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

public void shouldExecuteIterationsConcurrently() throws InterruptedException {

    List<Thread> threads = new ArrayList<>();
    for (int user = 1; user <= USERS; user++) {
        Thread thread = new Thread(new UserFlow(user));
        thread.start();
        threads.add(thread);
    }
    // 종료 조건 - 가장 효율적인 방법은 아니지만 의도대로 동작한다.
    for (Thread thread : threads) {
        thread.join();
    }
}

static class UserFlow implements Runnable {

    private final int user;
    private final List<String> serviceResult = new ArrayList<>();

    UserFlow(int user) {
        this.user = user;
    }

    @SneakyThrows
    @Override
    public void run() {
        Thread threadA = new Thread(new Service(this, "A", SERVICE_A_LATENCY, user));
        Thread threadB = new Thread(new Service(this, "B", SERVICE_B_LATENCY, user));
        threadA.start();
        threadB.start();
        threadA.join();
        threadB.join();

        List<Thread> threads = new ArrayList<>();
        for (int i = 1; i <= PERSISTENCE_FORK_FACTOR; i++) {
            Thread thread = new Thread(new Persistence(i, serviceResult.get(0), serviceResult.get(1)));
            thread.start();
            threads.add(thread);
        }
        
        // 종료 조건 - 가장 효율적인 방법은 아니지만 의도대로 동작한다.
        for (Thread thread : threads) {
            thread.join();
        }
    }

    public synchronized void addToResult(String result) {
        serviceResult.add(result);
    }
}
// Service와 Persistence 구현 코드는 생략

재미있는 사실

스레드를 무한정 생성할 수는 없다. OS마다 다르며 필자의 64비트 시스템에서 스레드 하나 당 1MB의 메모리(스레드 스택에 사용되는 메모리)를 점유한다. 요청 1000개, 30회 저장으로 설정하고 실행하면 33,000개의 스레드를 생성하려다가 Out Of Memory 에러가 발생한다.

ExecutorService

자바 5에 ExecutorService가 도입됐다. 스레드 풀링을 통해 새 스레드 생성 부담을 덜고 스레드를 로우 레벨로 다루는 부담을 덜어내는 게 주된 목표였다. 태스크는 ExecutorService에 submit 되고, 큐에 들어간다. 작업 가능한 스레드가 큐에서 태스크를 가져가서 실행한다.

눈여겨 볼 점은 다음과 같다.

• JVM 스레드 갯수가 여전히 OS 스레드 갯수에 의해 제한을 받는다.
• 스레드 풀에서 스레드를 하나 가져가면, 그 스레드는 연산을 수행하지 않더라도 다른 곳에 사용되지 못하고 낭비된다.
• Future가 반환되므로 발전된 것 같아 보이지만 조립(compose)할 수 없으며, 반환값을 얻기 위해 get()을 호출하면 태스크가 완료될 때까지 블로킹 된다.

코드

대체로 앞에서 살펴본 ‘네이티브 멀티 스레딩’과 비슷하다. 가장 큰 차이점은 스레드를 직접 생성하지 않고, 태스크를 ExecutorService에 submit 한다는 점이다. 스레드 풀 생성과 관리는 ExecutorService가 담당한다.

또 다른 점은 서비스 A와 서비스 B를 동기화하기 위해 join()을 사용하지 않는다는 점이다. 대신에 반환되는 Future의 get()을 호출해서 값이 반환될 때까지 블로킹한다.

예제에서는 2,000개의 스레드를 가진 스레드 풀이 사용됐다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

public void shouldExecuteIterationsConcurrently() throws InterruptedException {

    for (int user = 1; user <= USERS; user++) {
        executor.execute(new UserFlow(user));
    }

    // 종료 조건
    latch.await();
    executor.shutdown();
    executor.awaitTermination(60, TimeUnit.SECONDS);
}

static class UserFlow implements Runnable {

    private final int user;

    UserFlow(int user) {
        this.user = user;
    }

    @SneakyThrows
    @Override
    public void run() {
        Future<String> serviceA = executor.submit(new Service("A", SERVICE_A_LATENCY, user));
        Future<String> serviceB = executor.submit(new Service("B", SERVICE_B_LATENCY, user));

        for (int i = 1; i <= PERSISTENCE_FORK_FACTOR; i++) {
            executor.execute(new Persistence(i, serviceA.get(), serviceB.get()));
        }

        latch.countDown();
    }
}
// Service와 Persistence 구현 코드는 생략

재미있는 사실

2천 개의 스레드를 사용했을 때 앞서 살펴본 ‘네이티브 멀티 스레딩’에 비해 훨씬 나은 성능을 보이고 있다.

스레드 풀 크기를 제대로 설정하지 않으면 데드락이 발생하기도 한다. 풀 크기를 예를 들어 10정도로 작게 잡고 요청 갯수와 저장 횟수를 늘리면 데드락 때문에 아무 것도 제대로 완료되지 않는 것을 볼 수 있다. 왜냐하면 하나의 UserFlow가 다수의 스레드를 필요로 하도록 구현돼 있고, 다수의 UserFlow를 ExecutorService에 submit 하므로 풀에 있는 모든 스레드를 점유하게 되어 태스크를 완료할 수 있는 스레드가 남아있지 않기 때문이다.

예제에서는 스레드 풀 크기가 고정된 fixedThreadPool을 사용했지만, 크기가 동적으로 변하는 다른 스레드 풀을 사용할 수도 있다.

Fork/Join 프레임워크

자바 7에서 ExecutorService 기반으로 만들어진 Fork/Join 프레임워크가 도입됐다. Fork/Join 프레임워크는 재귀적으로 더 작은 크기로 쪼갤 수 있는 태스크를 효율적으로 처리하기 위해 만들어졌다. Fork/Join 프레임워크가 ExecutorService를 대체할 거라는 기대도 있었지만, Fork/Join 프레임워크는 동시 실행에 대해 개발자가 제어할 수 있는 옵션이 더 적으므로 ExecutorService는 여전히 계속 사용되고 있다.

ExecutorService와 확연히 다른 점은 작업 빼가기(work-stealing)다.

스레드 풀에 있던 스레드 A에 과부하가 걸려서 A 스레드 내부 큐가 꽉 차 있을 때, 스레드 풀에 있는 다른 스레드 B가 ExecutorService의 메인 큐에 있는 태스크를 가져오는 대신에 과부하 걸린 스레드 A 내부 큐에 있는 태스크를 가져와서 처리할 수 있다.

코드

스트림과 람다식 덕분이기도 하지만, 코드가 전체적으로 점점 짧아지고 있다.

Fork/Join 풀에 태스크를 submit 할 수 있는 UserFlowRecursiveAction를 구현했다. 예제 태스크가 재귀적으로 분할될 성질이 아니기 때문에 이런 태스크에 Fork/Join 프레임워크를 사용하는 것은 사실 적합하지는 않다. 다만 앞서 다뤄온 예제들의 연장선상에서 Fork/Join 프레임워크가 어떻게 동작하는지 알아보자는 목적에는 부합한다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

public void shouldExecuteIterationsConcurrently() throws InterruptedException {

    commonPool.submit(new UserFlowRecursiveAction(IntStream.rangeClosed(1, USERS)
            .boxed()
            .collect(Collectors.toList())));

    // Stop Condition
    commonPool.shutdown();
    commonPool.awaitTermination(60, TimeUnit.SECONDS);
}

public static class UserFlowRecursiveAction extends RecursiveAction {

    private final List<Integer> workload;

    public UserFlowRecursiveAction(List<Integer> workload) {
        this.workload = workload;
    }

    @Override
    protected void compute() {

        if (workload.size() > 1) {
            commonPool.submit(new UserFlowRecursiveAction(workload.subList(1, workload.size())));
        }

        int user = workload.get(0);

        ForkJoinTask<String> taskA = commonPool.submit(() -> service("A", SERVICE_A_LATENCY, user));
        ForkJoinTask<String> taskB = commonPool.submit(() -> service("B", SERVICE_B_LATENCY, user));

        IntStream.rangeClosed(1, PERSISTENCE_FORK_FACTOR)
                .forEach(i -> commonPool.submit(() -> persistence(i, taskA.join(), taskB.join())));
    }
}

재미있는 사실

앞서 ‘ExecutorService’ 단원에서 해봤던 것처럼 이번에도 풀 크기를 10 정도로 잡게 잡고 실행해보자. 이번에는 Fork/Join 프레임워크의 작업 빼가기 기능 덕분에 데드락이 발생하지 완료된다. 하지만 Fork/Join 프레임워크를 사용한다고해서 데드락이 항상 발생하지 않는 것은 아니다. Fork/Join 프레임워크에서는 태스크가 어떤 방식으로 더 작은 태스크로 분할될 수 있는지에 따라 데드락 발생 여부가 정해진다.

예제 태스크는 재귀적 분할에 딱 들어맞는 태스크가 아니며, Fork/Join 프레임워크에 추가된 로직 때문에 실행 성능은 좋지 않아서 ExecutorService 방식 보다도 오히려 떨어진다. 하지만 작은 풀 사이즈에서도 데드락이 발생하지 않았으므로 안정성은 더 높다.

CompletableFuture

자바 8에서 도입된 CompletableFuture는 Fork/Join 프레임워크를 기반으로 만들어졌다. 연산 결과를 모아서(combine) 처리할 수 있는 메서드가 하나도 없었고, 에러 처리를 위한 방법도 없었던 Future 인터페이스 도입 이후로 오랫동안 기다려 왔던 진화가 CompletableFuture에서 드디어 이루어졌다.

CompletableFuture를 통해 개선된 점은 다음과 같다.

• 더 개선된 함수형 프로그래밍 스타일 도입
• 로직을 조립(compose)하고, 결과를 모아서 처리(combine)하고, 비동기 연산 과정을 실행하고, 에러를 처리할 수 있는 50여개의 메서드 추가
• CompletableFuture의 평문형 API(fluent API) 대부분은 뒤에 Async 접미사가 붙은 것과 붙지 않은 것, 이렇게 2가지씩 짝지어져 있다. Async 접미사가 붙은 메서드는 해당 연산을 다른 스레드에서 실행하려고 할 때 사용된다.

코드

앞에서 다룬 ‘Fork/Join 프레임워크’보다도 훨씬 더 짧아졌고 압축적이다.

분량뿐 아니라 코딩 스타일 자체에서도 주목할만한 패러다임 변화가 눈에 띈다. 비동기 실행 결과를 모아서 처리하는 작업이 훨씬 자연스러운 코드로 표현된다.

하지만 함수형 프로그래밍 스타일에 익숙하지 않은 사람들에게는 대단히 생소해 보일 수도 있으며, 금방 적응하기 어려울 수도 있다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

public void shouldExecuteIterationsConcurrently() throws InterruptedException, ExecutionException {
    CompletableFuture.allOf(IntStream.rangeClosed(1, USERS)
            .boxed()
            .map(this::userFlow)
            .toArray(CompletableFuture[]::new)
    ).get();
}

@SneakyThrows
private CompletableFuture<String> userFlow(int user) {
    return CompletableFuture.supplyAsync(() -> serviceA(user), commonPool)
            .thenCombine(CompletableFuture.supplyAsync(() -> serviceB(user), commonPool), this::persist);
}

@SneakyThrows
private String persist(String serviceA, String serviceB) {
    CompletableFuture.allOf(IntStream.rangeClosed(1, PERSISTENCE_FORK_FACTOR)
            .boxed()
            .map(iteration -> CompletableFuture.runAsync(() -> persistence(iteration, serviceA, serviceB), commonPool))
            .toArray(CompletableFuture[]::new)
    ).join();

    return "";
}

재미있는 사실

CompletableFuture가 Fork/Join 프레임워크를 바탕으로 만들어졌음에도 불구하고 실행 성능은 훨씬 좋다.

CompletableFuture는 자바에 있는 몇 안 되는 모나드(monad) 중 하나다. 모나드를 알아보려면 이상한 나라의 앨리스에 나오는 토끼굴에 들어가는 모험을 감수해야 하므로 이 글에서는 다루지 않는다.

Reactive

리액티브 얘기를 이어가지 전에 먼저 반드시 구분해둬야 할 것이 있다. 리액티브 아키텍처와 리액티브 프로그래밍은 완전히 다르다는 점이다. 이 글에서는 비동기 데이터 스트림을 처리하고 에러 처리와 배압(backpressure)를 확고하게 지원하는 리액티브 프로그래밍만을 다룬다.

CompletableFuture가 진화해서 리액티브 방식이 됐다고 볼 수도 있지만 사실은 물론 그 이상이다. 리액티브 프로그래밍의 주요 목표는 프로그램 구조를 비동기 이벤트 스트림으로 재구성하는 것이며, 스레드 관리는 라이브러리/프레임워크에 위임한다.

주목할 점은 다음과 같다.

• 데이터를 발생시키는 Observable, 데이터를 소비하는 Observer, 스레드를 관리하는 Scheduler의 삼위 일체
• 리액티브 방식을 도입하면 프로그램 흐름 전부가 리액티브 방식으로 같이 바뀌어야 한다는 점에서 전염성이 강하다. 일부에 블로킹 코드가 남아 있으면 리액티브의 장점은 전혀 발휘되지 못한다.
• 리액티브 구현체도 여러가지가 있다. 처음에는 RxJava가 있었지만 최근에는 스프링의 Reactor가 대세다. 액터 모델을 구현하는 Akka 프레임워크는 RxJava나 Reactor보다 더 급진적인 리액티브 프로그래밍을 적용하고 있다.

코드

리액티브는 이 글에서는 유일하게 자바 언어 자체적으로는 제공되지 않는 솔루션이다. 예제에서는 스프링 Reactor를 사용했지만 RxJava도 크게 다르지 않다.

명령형 코딩 스타일에만 익숙하다면 이제 리액티브 코드가 아주 생소해 보일 수 있다. 리액티브 코드에 익숙해지려면 단순히 코딩뿐 아니라 테스트와 디버깅에서도 마인드셋을 바꿔야 한다. 특히 리액티브 코드의 테스트와 디버깅은 상당히 어렵지만 충분히 투자해볼만 하다.

리액티브 코딩 스타일을 마스터하면 지속적으로 발전하는 리액티브 지원 라이브러리의 도움에 힘입어 대단히 성능이 좋은 코드를 효율적으로 작성할 수 있다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

public void shouldExecuteIterationsConcurrently() {
    Flux.range(1, USERS)
            .flatMap(i -> Mono.defer(() -> userFlow(i)).subscribeOn(Schedulers.parallel()))
            .blockLast();
}

private Mono<String> userFlow(int user) {

    Mono<String> serviceA = Mono.defer(() -> Mono.just(serviceA(user))).subscribeOn(Schedulers.elastic());
    Mono<String> serviceB = Mono.defer(() -> Mono.just(serviceB(user))).subscribeOn(Schedulers.elastic());

    return serviceA.zipWith(serviceB, (sA, sB) -> Flux.range(1, PERSISTENCE_FORK_FACTOR)
            .flatMap(i ->
                    Mono.defer(() -> Mono.just(persistence(i, sA, sB))).subscribeOn(Schedulers.elastic())
            )
            .blockLast()
    );
}

재미있는 사실

작업 처리를 더 세밀하게 제어할 수 있게 됐지만, 스레드 풀 튜닝은 알아서 수행되므로 개발자가 직접 할 필요는 없다. 예제 코드에서는 리액티브의 진정한 장점 중 하나인 에러 처리나 배압 처리는 사용되지 않았다.

실행 성능은 CompletableFuture에 비해 떨어진다.

Project Loom

프로젝트 룸(Loom)은 아직 정식 출시되지 않았다. 2020년 12월 현재 기준 자바 16 Early Access 버전에 포함돼서 이것저것 실제로 실험해볼 수는 있지만 자바 16에 포함될지는 미지수다.

프로젝트 룸은 가상 스레드(Virtual Thread)와 관련된 여러 기능이 포함돼 있다. 그 중에서 ‘가상 스레드’와 ‘구조적 동시성(Structured Concurrency)’만 이 글에서 다룬다.

가상 스레드를 사용하면 JVM 스레드 : OS 스레드 = 1 : 1라는 오랜 등식이 더이상 성립되지 않는다. 가상 스레드는 기존의 JVM 스레드에 비해 훨씬 가볍고 저렴하다. 구조적 동시성을 도입하면 스레드 라이프타임이 스레드가 사용된 코드 블록과 연관(correlate)돼서 동기화가 훨씬 분명해지고, 우리가 익숙한 명령형 코딩 스타일을 그대로 사용할 수 있다.

주목해볼 점은 다음과 같다.

• 메타데이터, 스택 메모리, 컨텍스트 스위치 시간이 네이티브 OS 스레드의 수 분의 일 밖에 되지 않을만큼 가볍다.
• 아직 지원 도구가 충분하지 않다. 프로젝트 룸을 사용할 때 IntelliJ, Gradle, Lombok, JProfiler 등에서 여러가지 이슈가 발생했다. 그래서 프로젝트 룸 예제 코드는 별도의 프로젝트로 따로 작성했다.

코드

동시성을 전혀 사용하지 않은 단순하고 쉬운 코드와는 차이가 좀 있지만, 동시성을 사용했던 다른 코드들보다는 훨씬 단순하고 친숙해보인다.

가상 스레드를 지원하는 새로운 ExecutorService 구현체가 있고, AutoCloseable 인터페이스를 구현하고 있어서 try-with-resource 구문으로 사용해서 안전하게 자원을 열고 닫을 수 있다.

구조적 동시성이 적용돼 있어서 부모 스레드는 자기가 생성한 모든 자식 스레드의 종료를 try 블록 안에서 기다린다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

@SneakyThrows
private void startConcurrency() {
    try (var e = Executors.newVirtualThreadExecutor()) {
        IntStream.rangeClosed(1, USERS).forEach(i -> e.submit(() -> userFlow(i)));
    }
}

@SneakyThrows
private void userFlow(int user) {
    List<Future<String>> result;
    try (var e = Executors.newVirtualThreadExecutor()) {
        result = e.invokeAll(List.of(() -> serviceA(user),() -> serviceB(user)));
    }

    persist(result.get(0).get(), result.get(1).get());
}

private void persist(String serviceA, String serviceB) {
    try (var e = Executors.newVirtualThreadExecutor()) {
        IntStream.rangeClosed(1, PERSISTENCE_FORK_FACTOR)
                .forEach(i -> e.submit(() -> persistence(i, serviceA, serviceB)));
    }
}

재미있는 사실

JVM 차원에서의 개선이기 때문에 도입되면 많은 레거시 애플리케이션/라이브러리들이 별다른 수정 없이도 성능 개선 효과를 그대로 누릴 수 있다.

표에서 나타난 것처럼 실제로 몇 개의 OS 스레드를 사용했는지는 알 수 없지만, OS 스레드 갯수가 중요한 것은 아니다.

중요한 것은 프로젝트 룸을 사용했을 때 예제 실행 총 소요 시간이 이상적인 수치인 1,300ms에 가장 가깝다는 점이다. 요청 갯수나 저장 횟수를 늘리더라도 앞에서 살펴봤던 다른 방식들처럼 뚜렷한 성능 저하를 보이지도 않는다. 따라서 프로젝트 룸의 확장성이 상당히 좋다고 얘기할 수 있다.

결론

동시성 처리가 복잡하다는 건 분명한 사실이다. 동시성을 적용하지 않았던 ‘동시성 미사용’ 예제에 비해 ‘네이티브 멀티 스레딩’ 예제 코드는 훨씬 복잡하고 읽기도 어려우며 디버그하기는 거의 불가능했다.

하지만 동시성 처리 전문가가 되고 싶지 않거나 되고 싶더라도 시간이 없었던 개발자들에게도 마침내 희망의 등불이 켜졌다. 프로젝트 룸 덕분에 동시성 처리 전문가가 되지 않더라도 충분히 성능 좋은 코드를 작성할 수 있게 됐다.

프로젝트 룸이 출시되면 리액티브 프로그래밍을 완전히 대체할지는 아직 알 수 없다. 만병통치약은 존재하지 않는다는 사실을 감안할 때 프로젝트 룸과 리액티브 프로그래밍은 공존할 가능성이 높다고 본다. 다양한 지원 도구를 갖고 있는 리액티브 프로그래밍으로 해결하는 것이 더 나은 문제도 있을 것이다. 아직까지는 지원 도구가 많지 않지만 자바 동시성 진화 과정에서 더 자연스럽고 더 친숙한 다음 단계는 프로젝트 룸이라고 할 수 있다.

예제 성능 측정치는 모두 필자의 로컬 장비에서 수행된 결과이므로 과학적인 검증을 거친 측정치라고 볼 수는 없다. 하지만 다양한 방식을 비교하는 목적으로는 의미있는 수치라고 할 수 있다.

프로젝트 룸 적용 코드는 여기에서, 룸을 사용하지 않은 다른 코드는 여기에서 확인할 수 있다.

코딩 호러의 글 The Ten Commandments of Egoless Programming을 무부심 프로그래밍 십계명이라는 이름으로 옮겨봤다.

무부심이라는 표현은 법정 스님의 무소유에서 따왔다.
무소유는 소유하지 말자가 아니라, 불필요한 소유에서 벗어나자는 것이다.

마찬가지로 자부심을 갖지 말자가 아니라, 불필요하게 지나친 자부심에서 벗어나자는 뜻에서,
Egoless를 무부심으로 옮겼다.

1. 실수한다는 점을 이해하고 받아들여라

실수하지 않을 수는 없다. 실수가 실제 운영 코드에 스며들기 전에 찾아내는 것이 핵심이다.
다행스럽게도 극히 일부를 제외한 우리 대부분은 실수하면서 일을 한다.
그러니 우리는 실수에서 배우고, 웃고, 털어버리고 한 걸음 더 내디디면 된다.

2. 너 자신과 너의 코드를 구별할 줄 알라

리뷰의 목적은 오직 문제 발견 뿐이다. 그리고 문제가 있다면 발견이 된다.
내가 짠 코드에서 문제가 발견된다면 그저 문제가 있는 코드를 발견했을 뿐이다.
그러니 내가 짠 코드에 숨어있던 문제를 나 자신에 대한 문제라고 생각하지 말자.
이건 내가 아니라 동료에 대해서도 마찬가지다.

3. 고수는 분명히 존재한다

물어보면 고수는 답을 해줄 수 있다. 그러니 다른 사람들에게 묻고 받아들여라.
특히 물어볼 필요가 없다고 생각할 때에도 묻고 받아들여라.

4. 의논 없이 문제 없는 코드를 고치지 마라

잘못된 코드를 고치는 것과 잘 실행되는 코드를 다른 스타일로 재작성하는 것은 미세하게 다르다.
그 차이를 이해하고, 코드 스타일은 혼자서 강제하지 말고 코드 리뷰를 거친 후에 바꾸자.

5. 잘 모르는 사람도 존중, 존경, 인내를 갖고 대하라

개발자와 정기적으로 협업하는 대부분의 비개발자는, 개발자는 좋게 말하면 자기가 무척 잘난 줄 착각하는 사람이고, 나쁘게 말하면 울보 어린아이 같다고 생각한다.
참을성 없이 걸핏하면 화를 내면서 이런 고정관념을 고착화하는데 일조하지 말라.

6. 세상에서 변하지 않는 유일한 사실은, 변한다는 사실뿐이다

변화에 마음을 열고 웃으면서 받아들여라.
요구사항, 플랫폼, 도구 등 모든 것은 변하기 마련이다. 변화를 싸워서 물리쳐야할 대상으로 생각하지 말고 새로운 도전이라고 생각하라.

7. 진정한 권위는 지위가 아니라 지식에서 나온다

지식에는 권위가 따라오고, 권위에는 존경이 따라온다. 따라서 무부심 세상에서 존경을 받으려면 지식을 연마하라.

8. 신념을 위해 싸우되 패배도 받아들여라

너의 의견은 거절될 수도 있다. 나중에 너의 의견이 옳았다고 판명되더라도 복수하거나 ‘거봐 내 말대로 했어야지’ 같은 얘기를 여러 번 하지 마라.
아깝게 버려진 너의 의견을 순교자인 것처럼 미화하거나 강령으로 내세우지 마라.

9. 골방 개발자가 되지 마라

콜라를 살 때만 방 밖으로 나오는 골방 개발자가 되지 마라. 개방적으로 협력하는 세상에는 연락도 안 되고, 보이지도 않고, 제어할 수도 없는 골방 개발자를 위한 자리는 없다.

10. 사람이 아니라 코드를 비평하라 - 코드에게 친절하지 말고 코드를 작성한 사람에게 친절하라

가능한 모든 의견을 긍정적으로 표현하고 코드를 개선하는 방향으로 전개하라. 현재 상황에서 정해진 표준, 규격, 성능 개선 등을 바탕으로 의견을 개진하라.

오늘 일하다 의견을 나누게 된 Constants와 Util 얘기

애플리케이션이 실행되는 서버 IP를 가져와서 로그에 남길 일이 있다.

이유는 모르겠지만 InetAddress.getLocalHost().getHostAddress()로 찍으면 127.0.0.1 이 출력돼서 다른 방법을 찾아보니 아래와 같이 NetworkInterface 라는 놈을 써서 구할 수 있었다.

로직이 복잡해 보이긴 하지만 어쨌든 서버 IP라는 상수를 구하는 로직이라 아래와 같이 Constants 클래스에 private static 메서드로 넣었는데,

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

public class Constants {
    // 다른 상수 2개 있음

    public static final String SERVER_IP = getServerIp();

    private static String getServerIp() {
        try {
            Enumeration<NetworkInterface> enis = NetworkInterface.getNetworkInterfaces();
            List<NetworkInterface> nis = Collections.list(enis);
            for (NetworkInterface ni : nis) {
                if (ni.isUp() && !ni.isLoopback() && ni.getHardwareAddress() != null) {
                    List<InterfaceAddress> addresses = ni.getInterfaceAddresses();
                    for (InterfaceAddress address : addresses) {
                        if (!StringUtils.isEmpty(address.toString())) {
                            return address.getAddress().getHostAddress();
                        }
                    }
                }
            }
        } catch (SocketException e) {
            // 없으면 공백 반환 처리
        }
        return "";
    }
}

동료가 Constants 에는 메서드가 있으면 Constants 클래스 성격에 안 맞는다며 getServerIp()를 외부 Util 클래스로 빼자고 한다.

엉? 아니 Constants 클래스에 메서드가 있으면 안 되는 건가? 다른 메서드도 아니고 상수를 구하는 데 사용되는 private static 메서드인데?

나는 이 메서드는 분명히 SERVER_IP 라는 상수를 구하는 책임을 가지고 있고, 외부에서 호출할 필요가 없으므로 외부로 뺄 게 아니라 private static 으로 Constants 에 두는 게 적절하다고 봤다. 이게 정보와 로직을 함께 모아 넣자는 캡슐화의 원리에 부합하기 때문이다.

물론 상수가 많아지면 상황에 맞게 클래스를 분리해서 비대화를 막아야겠지만, 지금은 비대화와는 거리가 멀어 보이고, 분리한다고 해도 메서드의 존재 여부가 분리의 기준일 리는 없다고 생각한다.

하지만 동료는 Constants 라는 이름의 클래스는 통상적으로 상수만 모아 놓는 목적으로 만드는 거라, 이 클래스에 메서드를 넣으면 그 목적을 잃는다고 한다. 다른 동료도 이 메서드는 Util 클래스로 빼는 게 좋다고 한다.

아 그래? 메서드를 쓰면 안 되는 건가? 하고 다음과 같이 복잡해보이지만 메서드 호출이 아닌 단순 할당식으로 바꿔서 반응을 살펴봤다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

public static final String SERVER_IP = ((Supplier<String>) () -> {
     try {
         Enumeration<NetworkInterface> enis = NetworkInterface.getNetworkInterfaces();
         List<NetworkInterface> nis = Collections.list(enis);
         for (NetworkInterface ni : nis) {
             if (ni.isUp() && !ni.isLoopback() && ni.getHardwareAddress() != null) {
                 List<InterfaceAddress> addresses = ni.getInterfaceAddresses();
                 for (InterfaceAddress address : addresses) {
                     if (!StringUtils.isEmpty(address.toString())) {
                         return address.getAddress().getHostAddress();
                     }
                 }
             }
         }
     } catch (SocketException e) {
         // 없으면 공백 반환 처리
     }
     return "";
 }).get();

예상한대로 여전히 복잡한 로직이 있으니 외부 Util 로 빼는 게 좋다고 한다.

그러니까 결국 메서드라는 형식이 문제가 아니라 상수를 구하는 로직이 복잡하다면 외부로 빼야한다는 얘기였던 거다. 요는 Constants 클래스에는 복잡한 로직을 두지 말고 오직 간단하고 단순한 상수만 넣자는 얘기다.

정리하면 다음과 같다.

SERVER_IP라는 똑같은 상수일지라도 그 값을 구하는 데,

• 로직이 동원된다면 Constants.SERVER_IP로 하지 말고, ServerUtil.SERVER_IP 로 해야 되고,
• 로직이 동원되지 않는다면 ServerUtil.SERVER_IP로 하지 말고, Constants.SERVER_IP로 해야 된다는 얘기

Util 이나 Constants 나 엎어치나 메치나 좌측 궁뎅이나 우측 방뎅이나 그게 그거 같긴 하지만,
설계라는 게 원래 책임, 역할, 이름 이런 거 고민하는 거니께, 탭이나 스페이스 보다는 유익한 얘기 아닐까?
(헉 실수다 감히 신성한 탭/스페이스를 운운하다니..=3=3)
여러분의 생각은?

참고로 실무적으로는 다른 동료의 의견에 따라 다음과 같이 마무리했다.

1

public static final String SERVER_IP = System.getProperty("java.rmi.server.hostname", "");

System.getProperty()라는 외부 클래스의 public static 메서드 호출은 허용되지만,
내부에 private static 메서드를 두고 호출하는 건 안 된다는 얘기다.

이 정도면 Constants 클래스에는 시각적으로 길어 보이는 로직을 두지 말자는 얘기이고,
이 지점부터는 설계나 구현 원리/원칙보다는 취향에 가까운 얘기라서,
더 이상 얘기를 나누면 생산적이지 않은 쓸모 없는 얘기로 흐를 가능성이 높으니 이 정도에서 멈췄다.

1편에서 블로킹 방식의 싱글 스레드 소켓 서버를 만들어봤고 다음의 문제가 있음을 발견했다.

• 블로킹 방식의 싱글 스레드 소켓 서버는 시간 끄는 이상한 클라이언트가 하나만 들어와도 서버가 먹통이 되고, 다른 클라이언트까지 먹통될 수 있다.

이제 시간 끄는 이상한 클라이언트가 들어오더라도 서버나 다른 클라이언트가 먹통이 되지 않도록 개선해야 한다.

가장 간단한 방법은 클라이언트의 요청마다 별개의 스레드에서 처리하게 하는 것이다. 가장 널리 사용하는 방식이며 서블릿도 이 방식에 기반을 두고 있고, 서블릿에 기반을 둔 Spring MVC도 이 방식이다.

Classic IO(BIO) - Multi Thread ServerSocket

서버에서 멀티 스레드를 사용한다면 크게 두 가지 전략이 떠오른다.

1. 멀티 스레드로 ServerSocket을 여러 개 띄우고, 이 스레드를 요청 처리 완료때까지 사용한다.
2. ServerSocket은 하나의 스레드로 하되, ServerSocket.accept()로 받아온 여러 소켓들을 여러 개의 스레드로 처리한다.

일단 1번은 불가다. 아래와 같은 코드는 두 번째 ServerSocket 생성 시

1
2

ServerSocket serverSocket1 = new ServerSocket(Constants.SERVER_PORT);
ServerSocket serverSocket2 = new ServerSocket(Constants.SERVER_PORT);

다음과 같이 BindException이 발생한다.

1

Exception in thread "main" java.net.BindException: Address already in use

따라서 사용할 수 있는 전략은 2번 뿐이다.

ServerSocket을 사용해서 서버 데몬을 만들고, accept()로 클라이언트의 연결 요청을 기다리는 것까지는 싱글 스레드 서버와 동일하다. 다만 들어온 연결 요청의 처리를 요청 마다 다른 스레드에서 담당한다는 것만 다르다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

public class EchoSocketServerMultiThread {

    public static void main(String[] args) throws IOException {
        EchoSocketServerMultiThread echoSocketServerMultiThread = new EchoSocketServerMultiThread();
        echoSocketServerMultiThread.start();
    }

    public void start() throws IOException {
        // 50개짜리 스레드 풀
        ExecutorService es = Utils.getCommonExecutorService(50);
        try (ServerSocket serverSocket = new ServerSocket(Constants.SERVER_PORT);
             FileOutputStream fos = Utils.getCommonFileOutputStream()
        ) {
            Utils.serverTimeStamp("===============================", fos);
            Utils.serverTimeStamp("Multi Thread Socket Echo Server 시작", fos);

            while (true) {
                Utils.serverTimeStamp("---------------------------", fos);
                Utils.serverTimeStamp("Echo Server 대기 중", fos);

                // accept() 는 연결 요청이 올 때까지 return 하지 않고 blocking
                Socket acceptedSocket = serverSocket.accept();

                // 연결 요청이 오면 새 thread 에서 요청 처리 로직 수행
                es.execute(() -> {
                    try {
                        Utils.serverTimeStamp("Client 접속!!!", fos);
                        Utils.serverTimeStamp("Echo 시작", fos);
//                    Utils.sleep(500L);
                        EchoProcessor.echo(acceptedSocket);
                        Utils.serverTimeStamp("Echo 완료", fos);
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    }
                });
            }
        }
    }
}

Executors를 이용해서 단순한 고정 크기 스레드 풀을 만들어 사용한다.

1
2
3
4
5
6
7
8

public abstract class Utils {

    // ...

    public static ExecutorService getCommonExecutorService(int nThreads) {
        return Executors.newFixedThreadPool(nThreads);
    }
}

실습

금수저 서버 + 금수저 스레드 풀

50개 짜리 스레드 풀을 사용하면서 1편에서 문제를 유발했던 시나리오 그대로 수행해보자.

1. EchoSocketServerMultiThread 실행
2. EchoSocketClient 실행, EchoSocketClient는 연결 요청 후 5초 후에 메시지를 서버에 전송
3. 5초 이내에 다른 터미널에서 echo -n '아무거나' | nc localhost 7777 실행해서 메아리가 터미널에 바로 찍히면 문제 해결

실제로는 실습 편의를 위해 EchoSocketClient가 5초가 아니라 1분 후에 보내도록 설정했다.
결과는 아래 움짤과 같이 여러 터미널 창에서 동시에 echo -n '아무거나' | nc localhost 7777를 실행해도 모두 거의 동시에 메아리가 출력됐다.

움짤 파일 크기를 작게하기 위해 4개의 터미널창만 캡쳐했지만 실제로는 12개의 창에서 동시에 요청을 날렸다. main 스레드 + EchoSocketClient 요청 처리 스레드 + 12개의 터미널 요청 처리 스레드, 총 14개의 스레드가 사용됐다.

EchoSocketClient 요청은 pool-1-thread-1 스레드가 담당했고, 12개의 터미널 요청은 pool-1-thread-2 ~ pool-1-thread-13 스레드가 각각 처리했다. 로그를 보면 쉽게 알 수 있다. <==로 표시한 부분은 설명이다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

scratchpad-server git:main 🍺🦑🍺🍕🍺 ❯ tail -f temp.log                                                                                               ✹
[SERVER -            main] 2020-11-02T01:36:54.427095 - ===============================
[SERVER -            main] 2020-11-02T01:36:54.446306 - Multi Thread Socket Echo Server 시작
[SERVER -            main] 2020-11-02T01:36:54.446730 - ---------------------------
[SERVER -            main] 2020-11-02T01:36:54.447015 - Echo Server 대기 중

[SERVER -            main] 2020-11-02T01:37:00.181527 - ---------------------------
[SERVER - pool-1-thread-1] 2020-11-02T01:37:00.181621 - Client 접속!!! <== 1분 후 메시지 보내는 EchoSocketClient 요청 처리 시작
[SERVER -            main] 2020-11-02T01:37:00.181954 - Echo Server 대기 중
[SERVER - pool-1-thread-1] 2020-11-02T01:37:00.181981 - Echo 시작
[CLIENT -            main] 2020-11-02T01:37:00.197423 - Client 시작

[SERVER -            main] 2020-11-02T01:37:08.412302 - ---------------------------
[SERVER - pool-1-thread-2] 2020-11-02T01:37:08.412373 - Client 접속!!! <== 여기서부터는 터미널 nc 클라이언트 요청 처리 시작
[SERVER - pool-1-thread-2] 2020-11-02T01:37:08.412695 - Echo 시작
[SERVER -            main] 2020-11-02T01:37:08.412671 - Echo Server 대기 중
[SERVER -            main] 2020-11-02T01:37:08.413172 - ---------------------------
[SERVER - pool-1-thread-3] 2020-11-02T01:37:08.413241 - Client 접속!!!
[SERVER - pool-1-thread-3] 2020-11-02T01:37:08.413521 - Echo 시작
[SERVER -            main] 2020-11-02T01:37:08.413473 - Echo Server 대기 중
[SERVER -            main] 2020-11-02T01:37:08.414077 - ---------------------------
[SERVER - pool-1-thread-4] 2020-11-02T01:37:08.414172 - Client 접속!!!
[SERVER - pool-1-thread-4] 2020-11-02T01:37:08.414427 - Echo 시작
[SERVER -            main] 2020-11-02T01:37:08.414355 - Echo Server 대기 중
[SERVER -            main] 2020-11-02T01:37:08.414899 - ---------------------------
[SERVER - pool-1-thread-5] 2020-11-02T01:37:08.415059 - Client 접속!!!
[SERVER -            main] 2020-11-02T01:37:08.415181 - Echo Server 대기 중
[SERVER - pool-1-thread-5] 2020-11-02T01:37:08.415326 - Echo 시작
[SERVER -            main] 2020-11-02T01:37:08.415654 - ---------------------------
[SERVER - pool-1-thread-6] 2020-11-02T01:37:08.415793 - Client 접속!!!
[SERVER -            main] 2020-11-02T01:37:08.415901 - Echo Server 대기 중
[SERVER - pool-1-thread-6] 2020-11-02T01:37:08.416132 - Echo 시작
[SERVER -            main] 2020-11-02T01:37:08.416741 - ---------------------------
[SERVER - pool-1-thread-7] 2020-11-02T01:37:08.417166 - Client 접속!!!
[SERVER -            main] 2020-11-02T01:37:08.417073 - Echo Server 대기 중
[SERVER - pool-1-thread-7] 2020-11-02T01:37:08.417448 - Echo 시작
[SERVER - pool-1-thread-8] 2020-11-02T01:37:08.417732 - Client 접속!!!
[SERVER - pool-1-thread-8] 2020-11-02T01:37:08.418018 - Echo 시작
[SERVER -            main] 2020-11-02T01:37:08.417644 - ---------------------------
[SERVER -            main] 2020-11-02T01:37:08.418353 - Echo Server 대기 중
[SERVER -            main] 2020-11-02T01:37:08.418818 - ---------------------------
[SERVER - pool-1-thread-9] 2020-11-02T01:37:08.418987 - Client 접속!!!
[SERVER -            main] 2020-11-02T01:37:08.419078 - Echo Server 대기 중
[SERVER - pool-1-thread-9] 2020-11-02T01:37:08.419201 - Echo 시작
[SERVER -            main] 2020-11-02T01:37:08.419664 - ---------------------------
[SERVER - pool-1-thread-10] 2020-11-02T01:37:08.419741 - Client 접속!!!
[SERVER -            main] 2020-11-02T01:37:08.419985 - Echo Server 대기 중
[SERVER - pool-1-thread-10] 2020-11-02T01:37:08.419997 - Echo 시작
[SERVER -            main] 2020-11-02T01:37:08.420454 - ---------------------------
[SERVER - pool-1-thread-11] 2020-11-02T01:37:08.421378 - Client 접속!!!
[SERVER -            main] 2020-11-02T01:37:08.421536 - Echo Server 대기 중
[SERVER - pool-1-thread-11] 2020-11-02T01:37:08.421598 - Echo 시작
[SERVER -            main] 2020-11-02T01:37:08.421942 - ---------------------------
[SERVER - pool-1-thread-12] 2020-11-02T01:37:08.422346 - Client 접속!!!
[SERVER -            main] 2020-11-02T01:37:08.422542 - Echo Server 대기 중
[SERVER - pool-1-thread-12] 2020-11-02T01:37:08.422558 - Echo 시작
[SERVER -            main] 2020-11-02T01:37:08.423231 - ---------------------------
[SERVER - pool-1-thread-13] 2020-11-02T01:37:08.423331 - Client 접속!!!
[SERVER -            main] 2020-11-02T01:37:08.423519 - Echo Server 대기 중
[SERVER - pool-1-thread-13] 2020-11-02T01:37:08.423555 - Echo 시작
[SERVER - pool-1-thread-12] 2020-11-02T01:37:08.440014 - Echo 완료
[SERVER - pool-1-thread-7] 2020-11-02T01:37:08.440467 - Echo 완료
[SERVER - pool-1-thread-4] 2020-11-02T01:37:08.440014 - Echo 완료
[SERVER - pool-1-thread-3] 2020-11-02T01:37:08.440820 - Echo 완료
[SERVER - pool-1-thread-13] 2020-11-02T01:37:08.441453 - Echo 완료
[SERVER - pool-1-thread-11] 2020-11-02T01:37:08.441606 - Echo 완료
[SERVER - pool-1-thread-9] 2020-11-02T01:37:08.442308 - Echo 완료
[SERVER - pool-1-thread-2] 2020-11-02T01:37:08.442538 - Echo 완료
[SERVER - pool-1-thread-6] 2020-11-02T01:37:08.443030 - Echo 완료
[SERVER - pool-1-thread-10] 2020-11-02T01:37:08.442771 - Echo 완료
[SERVER - pool-1-thread-8] 2020-11-02T01:37:08.443385 - Echo 완료
[SERVER - pool-1-thread-5] 2020-11-02T01:37:08.443648 - Echo 완료
[CLIENT -            main] 2020-11-02T01:38:00.227402 - 메시지 전송 시작 <== EchoSocketClient 가 1분 후 메시지 전송
[CLIENT -            main] 2020-11-02T01:38:00.238871 - 메시지 print 완료
[CLIENT -            main] 2020-11-02T01:38:00.242828 - 메시지 flush 완료
[CLIENT -            main] 2020-11-02T01:38:00.243060 - 서버 Echo 대기...
[CLIENT -            main] 2020-11-02T01:38:00.261370 - 서버 Echo 도착
[SERVER - pool-1-thread-1] 2020-11-02T01:38:00.261379 - Echo 완료 <== pool-1-thread-1은 클라이언트에 의해 1분 동안 블록돼 있는 동안 0.5초 지나가므로 클라이언트로부터 메시지 받자마자 Echo
[CLIENT -            main] 2020-11-02T01:38:00.263653 - 서버 Echo msg: Server Echo - 안녕, echo server

흑수저 서버 + 흑수저 스레드 풀

이번에는 메아리 처리에 0.5초가 걸리는 후진 서버에서 스레드도 4개만 사용해서 테스트해보자.

메아리 처리에 0.5초가 걸리도록 EchoProcessor에서 주석 처리 돼 있던 sleep()의 주석을 해제한다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

public abstract class EchoProcessor {

    private static final FileOutputStream fos = Utils.getCommonFileOutputStream();

    public static void echo(Socket socket) throws IOException {
        try (BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
             PrintWriter out = new PrintWriter(socket.getOutputStream())
        ) {
            String clientMessage = in.readLine();  // in에 읽을 게 들어올 때까지 blocking
            String serverMessage = "Server Echo - " + clientMessage + System.lineSeparator();
            Utils.sleep(500L);  // Echo 처리에 0.5초가 걸리도록 주석 해제
            out.println(serverMessage);
            out.flush();
        }
    }
}

EchoSocketServerMultiThread의 스레드 풀을 스레드 4개만 사용하도록 ExecutorService es = Utils.getCommonExecutorService(4);로 바꿔서 흑수저 스레드 풀을 만든다. 그리고 앞에서와 마찬가지로 테스트 해보면 아래 움짤처럼 터미널에 표시되는 메아리에 시차가 조금 발생하는 것을 확인할 수 있다.

요청 처리엔 시간이 필요하고, 스레드도 메모리 및 Context Switching 부담이 있어서 마냥 늘릴 수만은 없으므로, 이 흑수저 서버 + 흑수저 스레드 풀 시나리오가 현실에 존재하는 시나리오에 가깝다고 할 수 있다.

1분 지연 Java Socket Client 요청 1개, 12개의 터미널 nc 요청 처리 로그는 다음과 같다. 자세한 내용은 <==로 설명을 추가했다.
스레드 이름과 화살표 길이, 중간중간 0.5초 정도 차이나는 부분을 유의해서 살펴보면 이해하는 데 도움이 될 것이다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

scratchpad-server git:main 🍺🦑🍺🍕🍺 ❯ tail -f temp.log        
[SERVER -            main] 2020-11-02T12:29:29.653111 - ===============================
[SERVER -            main] 2020-11-02T12:29:29.675600 - Multi Thread Socket Echo Server 시작
[SERVER -            main] 2020-11-02T12:29:29.676223 - ---------------------------
[SERVER -            main] 2020-11-02T12:29:29.676571 - Echo Server 대기 중

[SERVER -            main] 2020-11-02T12:29:36.000378 - ---------------------------
[SERVER - pool-1-thread-1] 2020-11-02T12:29:36.000475 - Client 접속!!! <== 1분 후 메시지 보내는 EchoSocketClient 요청
[SERVER - pool-1-thread-1] 2020-11-02T12:29:36.000856 - Echo 시작
[SERVER -            main] 2020-11-02T12:29:36.000830 - Echo Server 대기 중
[CLIENT -            main] 2020-11-02T12:29:36.015713 - Client 시작
    
[SERVER -            main] 2020-11-02T12:29:43.072899 - ---------------------------
[SERVER - pool-1-thread-2] 2020-11-02T12:29:43.072977 - Client 접속!!! <== 터미널 요청 1 처리: 2번 스레드에서 시작
[SERVER -            main] 2020-11-02T12:29:43.073253 - Echo Server 대기 중 <== main 스레드는 블록되지 않고 계속 요청을 받아 스레드 풀에 전달하고 다음 요청 대기
[SERVER - pool-1-thread-2] 2020-11-02T12:29:43.073283 - Echo 시작 <== 요청 1 Echo 시작
[SERVER - pool-1-thread-3] 2020-11-02T12:29:43.073725 - Client 접속!!! <==== 터미널 요청 2 처리: 3번 스레드에서 시작
[SERVER -            main] 2020-11-02T12:29:43.073657 - ---------------------------
[SERVER - pool-1-thread-3] 2020-11-02T12:29:43.073938 - Echo 시작 <==== 요청 2 Echo 시작
[SERVER -            main] 2020-11-02T12:29:43.073984 - Echo Server 대기 중 <== main 스레드는 블록되지 않고 계속 요청을 받아 스레드 풀에 전달하고 다음 요청 대기
[SERVER -            main] 2020-11-02T12:29:43.074461 - ---------------------------
[SERVER - pool-1-thread-4] 2020-11-02T12:29:43.074550 - Client 접속!!! <====== 터미널 요청 3 처리: 4번 스레드에서 시작
[SERVER -            main] 2020-11-02T12:29:43.074694 - Echo Server 대기 중 <== main 스레드는 블록되지 않고 계속 요청을 받아 스레드 풀에 전달하고 다음 요청 대기
[SERVER - pool-1-thread-4] 2020-11-02T12:29:43.074716 - Echo 시작 <====== 요청 3 Echo 시작
[SERVER -            main] 2020-11-02T12:29:43.075061 - ---------------------------
[SERVER -            main] 2020-11-02T12:29:43.075294 - Echo Server 대기 중 <== main 스레드는 블록되지 않고 계속 요청을 받아 스레드 풀에 전달하고 다음 요청 대기
[SERVER -            main] 2020-11-02T12:29:43.075582 - ---------------------------
[SERVER -            main] 2020-11-02T12:29:43.075844 - Echo Server 대기 중
[SERVER -            main] 2020-11-02T12:29:43.076123 - ---------------------------
[SERVER -            main] 2020-11-02T12:29:43.076339 - Echo Server 대기 중
[SERVER -            main] 2020-11-02T12:29:43.076611 - ---------------------------
[SERVER -            main] 2020-11-02T12:29:43.076860 - Echo Server 대기 중
[SERVER -            main] 2020-11-02T12:29:43.080305 - ---------------------------
[SERVER -            main] 2020-11-02T12:29:43.080641 - Echo Server 대기 중
[SERVER -            main] 2020-11-02T12:29:43.080866 - ---------------------------
[SERVER -            main] 2020-11-02T12:29:43.081025 - Echo Server 대기 중
[SERVER -            main] 2020-11-02T12:29:43.081308 - ---------------------------
[SERVER -            main] 2020-11-02T12:29:43.081528 - Echo Server 대기 중
[SERVER -            main] 2020-11-02T12:29:43.081733 - ---------------------------
[SERVER -            main] 2020-11-02T12:29:43.081907 - Echo Server 대기 중
[SERVER -            main] 2020-11-02T12:29:43.082124 - ---------------------------
[SERVER -            main] 2020-11-02T12:29:43.082250 - Echo Server 대기 중 <== main 스레드는 블록되지 않고 계속 요청을 받아 스레드 풀에 전달하고 다음 요청 대기
[SERVER - pool-1-thread-2] 2020-11-02T12:29:43.593116 - Echo 완료 <== 요청 1 Echo 완료, 0.5초 소요
[SERVER - pool-1-thread-4] 2020-11-02T12:29:43.593116 - Echo 완료 <====== 요청 3 Echo 완료, 0.5초 소요
[SERVER - pool-1-thread-3] 2020-11-02T12:29:43.593413 - Echo 완료 <==== 요청 2 Echo 완료, 0.5초 소요
[SERVER - pool-1-thread-2] 2020-11-02T12:29:43.593569 - Client 접속!!! <== 요청 1 처리 완료 후 큐에 있던 후속 요청 A: 2번 스레드에서 처리 시작
[SERVER - pool-1-thread-4] 2020-11-02T12:29:43.593575 - Client 접속!!! <====== 요청 3 처리 완료 후 큐에 있던 후속 요청 C: 4번 스레드에서 처리 시작
[SERVER - pool-1-thread-2] 2020-11-02T12:29:43.593817 - Echo 시작 <== 요청 A
[SERVER - pool-1-thread-3] 2020-11-02T12:29:43.593683 - Client 접속!!! <==== 요청 2 처리 완료 후 큐에 있던 후속 요청 B: 3번 스레드에서 처리 시작
[SERVER - pool-1-thread-4] 2020-11-02T12:29:43.593889 - Echo 시작 <====== 요청 C
[SERVER - pool-1-thread-3] 2020-11-02T12:29:43.594157 - Echo 시작 <==== 요청 B
[SERVER - pool-1-thread-2] 2020-11-02T12:29:44.097963 - Echo 완료 <== 요청 A
[SERVER - pool-1-thread-4] 2020-11-02T12:29:44.097963 - Echo 완료 <====== 요청 C
[SERVER - pool-1-thread-2] 2020-11-02T12:29:44.098370 - Client 접속!!!
[SERVER - pool-1-thread-4] 2020-11-02T12:29:44.098411 - Client 접속!!!
[SERVER - pool-1-thread-4] 2020-11-02T12:29:44.098672 - Echo 시작
[SERVER - pool-1-thread-2] 2020-11-02T12:29:44.098608 - Echo 시작
[SERVER - pool-1-thread-3] 2020-11-02T12:29:44.097963 - Echo 완료 <==== 요청 B
[SERVER - pool-1-thread-3] 2020-11-02T12:29:44.099086 - Client 접속!!!
[SERVER - pool-1-thread-3] 2020-11-02T12:29:44.099283 - Echo 시작
[SERVER - pool-1-thread-4] 2020-11-02T12:29:44.603912 - Echo 완료
[SERVER - pool-1-thread-2] 2020-11-02T12:29:44.604235 - Echo 완료
[SERVER - pool-1-thread-4] 2020-11-02T12:29:44.604316 - Client 접속!!!
[SERVER - pool-1-thread-4] 2020-11-02T12:29:44.604478 - Echo 시작
[SERVER - pool-1-thread-2] 2020-11-02T12:29:44.604473 - Client 접속!!!
[SERVER - pool-1-thread-3] 2020-11-02T12:29:44.604637 - Echo 완료
[SERVER - pool-1-thread-2] 2020-11-02T12:29:44.604910 - Echo 시작
[SERVER - pool-1-thread-3] 2020-11-02T12:29:44.604962 - Client 접속!!!
[SERVER - pool-1-thread-3] 2020-11-02T12:29:44.605142 - Echo 시작
[SERVER - pool-1-thread-3] 2020-11-02T12:29:45.107464 - Echo 완료
[SERVER - pool-1-thread-4] 2020-11-02T12:29:45.107464 - Echo 완료
[SERVER - pool-1-thread-2] 2020-11-02T12:29:45.107464 - Echo 완료
[CLIENT -            main] 2020-11-02T12:30:36.051155 - 메시지 전송 시작 <== 1분 후 메시지 보내는 Java Socket Client
[CLIENT -            main] 2020-11-02T12:30:36.068386 - 메시지 print 완료
[CLIENT -            main] 2020-11-02T12:30:36.073274 - 메시지 flush 완료 <== 메시지 전송
[CLIENT -            main] 2020-11-02T12:30:36.073655 - 서버 Echo 대기...
[SERVER - pool-1-thread-1] 2020-11-02T12:30:36.581245 - Echo 완료 <== 메시지 전송 받은 후 0.5초 후에 Echo 완료
[CLIENT -            main] 2020-11-02T12:30:36.582077 - 서버 Echo 도착
[CLIENT -            main] 2020-11-02T12:30:36.585161 - 서버 Echo msg: Server Echo - 안녕, echo server

요는 다음과 같다.

• accept()는 요청이 오는 족족 스레드 풀에 전달
  • 요청을 받을 때까지는 블로킹이지만, 요청을 받은 후에는 블로킹하지 않고 바로 다음 요청을 블로킹하면서 대기
• 스레드 풀은
  • 사용 가능한 스레드가 있으면 요청을 바로 처리
  • 사용 가능한 스레드가 없으면 요청을 큐에 저장 후 나중에 처리

정리

• ServerSocket.accept()로 클라이언트의 요청을 받아들이는 건 하나의 스레드에서 담당하고,
• 요청의 처리는 요청마다 별도의 스레드에서 처리하게 하면,

• 시간을 오래 끄는 클라이언트가 일부 있더라도 서버와 나머지 클라이언트는 먹통이 발생하지 않는다.

• 요청을 블로킹 방식으로 처리하는 상황에서

  • 요청 처리에 일정 시간이 필요하고, 스레드가 충분하지 못하다면, 일부 요청은 큐에서 쌓여 대기하다가 시차를 두고 처리된다.
  • 큐에 쌓여 대기하는 요청을 줄이려면 스레드를 늘리는 수 밖에 없다.

결국 제한된 자원 상황에서 많은 요청을 처리하려면 스레드를 늘리는 수 밖에 없지만, 자체 스택을 갖고 있는 스레드를 많이 사용하면 메모리 한계에 부딪힐 수 있고, 많은 스레드는 많은 Context Switching 을 유발해서 성능 저하의 원인이 되기도 한다.

그렇다면 스레드를 무작정 늘리기보다는 스레드 사용 효율을 높이는 방법을 찾아봐야 할 것 같다.

3편에서는 스레드 효율을 높일 수 있는 NIO에 대해 알아본다.

서버 프로그래밍을 한다고는 하지만, 지난 수년 간 굴러도 스프링 위에서만 구르다보니 스프링 없이는, 아니 이제는 스프링만으로도 뭘 못할 것 같고 스프링 부트 없이는 간단한 메아리(Echo) 서버조차 못 만드는 경지지경에 이르렀다. 이 아니 부끄러운가..

그래서 Java가 제공해주는 classic IO, NIO, NIO2로 간단한 Echo Server를 만들어보면서 기본기를 좀 다져보려 한다.
만드는 데서 그치지 않고 그동안 간접 경험으로만 알아왔던 NIO, NIO2 의 장단점을 부하테스트를 통해 확인해보고자 한다.
나름 원대한 계획이지만 목표한 걸 모두 얻을 수 있을지는 미지수다. 그냥 달려보자.

Client

서버를 호출할 클라이언트는 크게 3가지다.

• Java Socket Client
• nc(netcat)
• JMeter Client

이 중에서 코딩이 필요한 건 Java Socket Client 뿐이고 코드는 다음과 같다. 이해를 위해 로깅을 많이 넣었는데, 로깅 빼면 설명할 것도 없다.
참고로 로깅을 콘솔이 아닌 temp.log 파일에 찍는다. 이유는 서버와 클라이언트의 로그를 한 군데 모아서 보는 게 이해하는 데 도움이 되기 때문이다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

package io.homo_efficio.server.socket;

import io.homo_efficio.server.common.Constants;
import io.homo_efficio.server.common.Utils;

import java.io.*;
import java.net.Socket;

/**
 * @author homo.efficio@gmail.com
 * created on 2020-10-10
 */
public class EchoSocketClient {

    public static void main(String[] args) throws IOException {
        String message = "안녕, echo server";

        try (Socket clientSocket = new Socket(Constants.SERVER_HOST_NAME, Constants.SERVER_PORT);
             FileOutputStream fos = Utils.getCommonFileOutputStream();
             PrintWriter out = new PrintWriter(clientSocket.getOutputStream());
             BufferedReader in = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()))
        ) {
            Utils.clientTimeStamp("Client 시작", fos);
            // Utils.sleep(5000L);  // 서버 blocking 확인 시 사용
            Utils.clientTimeStamp("메시지 전송 시작", fos);
            out.println(message);
            Utils.clientTimeStamp("메시지 print 완료", fos);
            out.flush();
            Utils.clientTimeStamp("메시지 flush 완료", fos);
            Utils.clientTimeStamp("서버 Echo 대기...", fos);
            // in.readLine() 은 읽을 데이터가 들어올 때까지 blocking 이므로 while (true) 불필요
            String messageFromServer = in.readLine();
            Utils.clientTimeStamp("서버 Echo 도착", fos);
            Utils.clientTimeStamp("서버 Echo msg: " + messageFromServer, fos);
        }
    }
}

Classic IO - Single Thread ServerSocket

이제 서버를 만들어 보자. 1번 타자는 Classic IO(또는 BIO(Blocking IO))로 만든 울트라 심플 싱글 스레드 소켓 서버다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

package io.homo_efficio.server.socket;

import io.homo_efficio.server.common.Constants;
import io.homo_efficio.server.common.EchoProcessor;
import io.homo_efficio.server.common.Utils;

import java.io.FileOutputStream;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;

/**
 * @author homo.efficio@gmail.com
 * created on 2020-10-10
 */
public class EchoSocketServerSingleThread {

    public static void main(String[] args) throws IOException {
        EchoSocketServerSingleThread echoSocketServerSingleThread = new EchoSocketServerSingleThread();
        echoSocketServerSingleThread.start();
    }

    public void start() throws IOException {
        try (ServerSocket serverSocket = new ServerSocket(Constants.SERVER_PORT);
             FileOutputStream fos = Utils.getCommonFileOutputStream()
        ) {
            Utils.serverTimeStamp("===============================", fos);
            Utils.serverTimeStamp("Echo Server 시작", fos);

            while (true) {
                Utils.serverTimeStamp("---------------------------", fos);
                Utils.serverTimeStamp("Single Thread Socket Echo Server 대기 중", fos);
                // accept() 는 연결 요청이 올 때까지 return 하지 않고 blocking
                Socket acceptedSocket = serverSocket.accept();

                // 연결 요청이 오면 accept() 가 반환하고 요청 처리 로직 수행
                Utils.serverTimeStamp("Client 접속!!!", fos);
//            Utils.sleep(50L);
                Utils.serverTimeStamp("Echo 시작", fos);
                EchoProcessor.echo(acceptedSocket);
                Utils.serverTimeStamp("Echo 완료", fos);
            }
        }
    }
}

ServerSocket으로 서버 소켓을 생성하고, accept()로 클라이언트의 연결을 기다리고, 연결이 오면 클라이언트에게 메시지를 메아리로 되돌려 준다.

메아리를 담당하는 EchoProcessor는 다음과 같다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

public abstract class EchoProcessor {

    private static final FileOutputStream fos = Utils.getCommonFileOutputStream();

    public static void echo(Socket socket) throws IOException {
        try (BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
             PrintWriter out = new PrintWriter(socket.getOutputStream())
        ) {
            String clientMessage = in.readLine();  // in에 읽을 게 들어올 때까지 blocking
            String serverMessage = "Server Echo - " + clientMessage + System.lineSeparator();
            out.println(serverMessage);
            out.flush();
        }
    }
}

Socket을 인자로 받고, 소켓에서 Reader, Writer를 뽑아내서, Reader에서 메아리를 읽고 ‘Server Echo -‘라는 문자열을 앞에 붙여서 Writer로 회신한다.

여기서 주의할 점이 있다. 서버가 보내는 메시지에 비어 있는 행이 포함돼야 클라이언트가 readLine()으로 읽을 때 행을 구별해서 문제 없이 읽고 출력할 수 있다. 비어 있는 행이 없으면 클라이언트의 readLine()이 계속 비어 있는 행을 기다리면서 서버와의 연결을 점유하게 되고, 싱글 스레드인 서버는 먹통 상태가 된다.

실습

1. EchoSocketServerSingleThread 를 실행하고, EchoSocketClient 를 실행하면 temp.log 파일에 다음과 같이 로그가 찍한다.

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16

   [SERVER - main] 2020-11-01T23:49:25.119684 - =============================== [SERVER - main] 2020-11-01T23:49:25.133603 - Echo Server 시작 [SERVER - main] 2020-11-01T23:49:25.133994 - --------------------------- [SERVER - main] 2020-11-01T23:49:25.134174 - Single Thread Socket Echo Server 대기 중 [SERVER - main] 2020-11-01T23:49:26.976560 - Client 접속!!! [SERVER - main] 2020-11-01T23:49:26.976861 - Echo 시작 [CLIENT - main] 2020-11-01T23:49:26.992329 - Client 시작 [CLIENT - main] 2020-11-01T23:49:27.006950 - 메시지 전송 시작 [CLIENT - main] 2020-11-01T23:49:27.007250 - 메시지 print 완료 [CLIENT - main] 2020-11-01T23:49:27.008839 - 메시지 flush 완료 [CLIENT - main] 2020-11-01T23:49:27.009160 - 서버 Echo 대기... [CLIENT - main] 2020-11-01T23:49:27.020318 - 서버 Echo 도착 [SERVER - main] 2020-11-01T23:49:27.021049 - Echo 완료 [SERVER - main] 2020-11-01T23:49:27.021302 - --------------------------- [SERVER - main] 2020-11-01T23:49:27.021471 - Single Thread Socket Echo Server 대기 중 [CLIENT - main] 2020-11-01T23:49:27.021674 - 서버 Echo msg: Server Echo - 안녕, echo server

2. 서버는 여전히 대기 중이므로 다른 터미널에서 echo -n '아무거나' | nc localhost 7777을 입력하면 다음과 같이 Echo 메시지가 바로 출력되어 나온다.

   1 2 3

   다른 터미널창 🍺🦑🍺🍕🍺 ❯ echo -n '아무거나' | nc localhost 7777 Server Echo - 아무거나

   1 2 3 4 5 6 7 8 9 10

   ... 윗 부분 생략 ... [SERVER - main] 2020-11-01T23:49:27.021302 - --------------------------- [SERVER - main] 2020-11-01T23:49:27.021471 - Single Thread Socket Echo Server 대기 중 [SERVER - main] 2020-11-02T00:03:47.863975 - Client 접속!!! [SERVER - main] 2020-11-02T00:03:47.874942 - Echo 시작 [SERVER - main] 2020-11-02T00:03:47.878276 - Echo 완료 [SERVER - main] 2020-11-02T00:03:47.878572 - --------------------------- [SERVER - main] 2020-11-02T00:03:47.878849 - Single Thread Socket Echo Server 대기 중

3. EchoSocketClient 에서 // Utils.sleep(5000L); // 서버 blocking 확인 시 사용라고 돼 있던 부분의 주석을 해제하고 실행해서 클라이언트가 서버와 연결된 후 5초 후에 서버에 메시지를 전송하도록 하고, 5초 안에 다른 터미널에서 echo -n '아무거나' | nc localhost 7777을 입력한다.

   • 그러면 메아리가 터미널에 금방 출력되지 않고 5초 후에 출력된다.
   • 이유는 앞서 말한 것처럼 EchoSocketClient가 5초 후에 메시지를 보내는 동안, EchoProcessor의 in.readLine()이 블로킹 상태로 대기하는데, 서버의 스레드도 1개 뿐이라 다른 요청을 accept() 할 수 없기 때문이다.
   • 그래서 터미널 클라이언트도 5초간 블로킹 상태로 대기하게 된다.
   • 결국 이상한 클라이언트가 하나 끼면 서버도 먹통되고 다른 클라이언트까지 먹통이 전파될 수 있다.

정리

• 블로킹 방식의 싱글 스레드 소켓 서버는 시간 끄는 이상한 클라이언트가 하나만 들어와도 서버가 먹통이 되고, 다른 클라이언트까지 먹통될 수 있다.

이 문제는 어떻게 해결할까? 2편에서 알아보자.

일반적으로 DB에 연결해서 어떤 작업을 할 때는 커넥션풀(Connection Pool)을 사용한다. DB 연결 자체가 비용이 많이 들기 때문에 미리 다수의 Connection 객체를 만들어서 풀에 넣어두고 필요할 때마다 꺼내쓰고 반납하기를 반복한다. 결국 응답 속도를 빠르게 하고 자원 효율성을 높이기 위해 커넥션풀을 사용한다.

커넥션풀을 사용하면 미리 만들어진 연결을 여러 곳에서 재사용하기 때문에 연결 객체에 어떤 공통의 상태를 두고 이를 변경해가면서 사용하면 안 된다. A라는 작업에서 필요에 의해 그 상태를 변경하면, A가 쓰고 반납한 연결 객체를 재사용하는 B라는 다른 작업에서는 A에 의해 변경된 상태에 의해 의도하지 않게 동작할 위험이 있기 때문이다.

배치 작업

하지만 배치(bach) 작업은 어떨까? 대부분의 배치 작업은 최종 사용자의 요청과는 무관하게 동작하며 따라서 최종 사용자에게 필요한 수준의 응답 속도를 필요로 하지 않는다. 그리고 보통 작업량도 많아서 DB 연결 생성 비용은 그 작업에 필요한 전체 비용과 비교하면 무시할 정도로 미미한 수준이다. 따라서 일반적인 배치 작업에서는 커넥션풀의 효용이 그다지 크지 않다.

게다가 작업 성격 상 커넥션별로 어떤 설정값을 변경해야 하는 경우도 많다. 예를 들어 Hive의 Dynamic Partition Insert을 사용할 때는 아래 set hive.exec.XXX=YYY와 같이 설정값을 변경해줘야 한다.

a, b로 설정한 게 c 실행 시 까지 유효해야 하는데, 실제로는 a, b, c 모두 서로 다른 커넥션에서 실행된다. 이는 크게 세 가지 치명적인 문제를 유발한다.

1. a, b 설정 내용은 해당 커넥션에 그대로 남아서 나중에 다른 작업에 영향을 미친다. 다른 작업은 의도한 것과 다르게 동작할 수 있다는 얘기다.
2. a, b 설정 내용은 c 실행 시에는 적용되지 않는다. 셋 모두 별개의 커넥션에서 실행되기 때문이다.
3. 1, 2에 의한 문제는 간헐적, 우발적으로 발생한다. 재연이 어렵고 디버깅이 어렵다.

참고로 a, b, c 실행 시 서로 다른 커넥션이 사용된다는 건 아래의 JdbcTemplate 구현 내용에서 확인할 수 있다.

위와 같이 결국 JdbcTemplate의 DB 작업 메서드가 실행될 때마다 DataSourceUtil.getConnection()로 커넥션을 가져오는데, 이게 결국 커넥션풀에서 그때그때 커넥션을 새로 가져온다.

배치 작업에서의 커넥션풀

앞에서 얘기한 것처럼 배치 작업에서의 커넥션풀의 효용은 크지 않다. 그리고 커넥션별 설정 변경 같은 사용 사례가 필요한 상황에서는 커넥션풀을 써서 커넥션을 재사용하는 것은 앞에서 얘기한 것처럼 해결이 어려운 문제를 유발하는 부작용만 떠안을 뿐이다.

그럼 커넥션풀을 안 쓰면 되는 거 아닌가? 맞다. 근데 실무적으로 편하게 사용할 수 있는 JdbcTemplate이 커넥션풀에서 커넥션을 가져오게 되어 있으므로 이 편리함을 그대로 유지하려면 커넥션풀을 사용해야 한다. 뭐야 그럼 어쩌라고?

커넥션풀을 사용하되 적당히 설정하면 커넥션을 사용할 때마다 커넥션 객체를 새로 생성하게 할 수 있다.

즉 커넥션풀을 사용하지만 일반적인 커넥션풀처럼 커넥션을 미리 만들어두고 재사용하는 게 아니라,

• 커넥션을 사용할 때마다 늘 새로 만들고,
• 사용 후에 바로 폐기되며,
• 동시 사용 커넥션의 최대 갯수만 제한을 두는

특별한 풀을 만들 수 있다.

아래는 Tomcat의 PoolProperties를 사용할 때의 설정값인데 다른 ConnectionPool 구현체를 쓰더라도 비슷한 설정방법이 있을 것이다.

1
2
3
4
5
6
7
8
9

driverClassName: a.b.c.d.GoodDriver
url: JDBC_URL
username: USERNAME
password: PASSWORD
initialSize: 0  # 풀 생성 시 커넥션 객체를 미리 생성하지 않음
maxActive: 100  # 동시 사용 가능 한 커넥션 객체의 갯수 상한값을 기본값인 100으로 명시
maxIdle: 0      # 사용 후 대기 중인 커넥션의 최대 갯수를 0으로 고정
minIdle: 0      # 사용 후 대기 중인 커넥션의 최소 갯수를 0으로 고정
minEvictableIdleTimeMillis: 0  # 사용 후 폐기 전 대기 시간을 0밀리초로 고정 -> 사용 후 바로 폐기

Tx 처리

위와 같이 커넥션풀을 설정하면 앞에서 살펴본 1번 문제는 해결할 수 있다. 하지만 a, b 설정이 c 실행시까지 유지되어야 하는 2번 문제는 어떻게 해결해야 할까?

하나의 Tx로 묶어주면 JdbcTemplate이 처음에 가져온 커넥션을 Tx 종료시까지 계속 사용할 수 있다. 스프링이라면 @Transactional을 사용하거나 아래와 같이 PlatformTransactionManager 를 사용해서 하나의 Tx로 묶어줄 수 있다.

1
2
3
4
5
6
7
8
9
10

TransactionStatus transactionStatus = this.transactionManager.getTransaction(new DefaultTransactionDefinition());

JdbcTemplate hiveTezJdbcTemplate = hiveTezDataReader.createJdbcTemplate();

// 아래 3개의 명령이 모두 동일한 커넥션 객체를 통해 실행됨
hiveTezJdbcTemplate.execute("set hive.exec.dynamic.partition.mode=nonstrict");
hiveTezJdbcTemplate.execute("set hive.exec.max.dynamic.partitions=50000");
hiveTezJdbcTemplate.execute(hiveCustomerConsumptionQuery);

this.transactionManager.commit(transactionStatus);

그림 출처: https://www.happycoders.eu/java/filechannel-bytebuffer-memory-mapped-file-locks/

NIO의 장점은 https://homoefficio.github.io/2016/08/06/Java-NIO는-생각만큼-non-blocking-하지-않다/ 를 참고하고, 여기에서는 FileChannel과 DirectBuffer 얘기만 다룬다.

ByteBuffer는 생성되는 위치를 기준으로 크게 나눠보면 JVM Heap 내에 생성되는 HeapByteBuffer와 JVM Heap 밖에 있는 Native 공간에 생성되는 DirectByteBuffer로 나눌 수 있다. 아래 그림에는 먼저 HeapByteBuffer와 MappedByteBuffer로 구분되는 걸로 보이는데 MappedByteBuffer도 Native 공간에 생성되며 파일 일부를 메모리에 매핑한다는 점 외에는 일반적인 direct byte buffer 와 동작이 다르지 않다고 API 문서에 나와있다.

위 그림에는 안 나와있지만 DirectByteBuffer는 DirectBuffer 인터페이스를 구현하고 있다.

HeapByteBuffer를 사용하면 JVM의 GC에 안전하게 의지할 수 있지만, CPU 개입 없이 I/O를 수행할 수 있고 불필요한 copy 부하가 발생하지 않아 성능적으로 유리한 점이 많은 DMA(DirectMemoryAccess)는 활용할 수 없다.

반대로 DirectByteBuffer를 사용하면 DMA의 혜택을 얻을 수 있지만, JVM의 GC를 벗어나게 되므로 메모리 관리 부담이 생겨난다.

그래서 FileChannel 을 사용할 때 상황에 맞게 HeapByteBuffer나 DirectByteBuffer 중에 골라서 쓰면 될 것 같..지만, 실상은 꼭 그렇지는 않다!

이제부터 코드와 함께 그 내부를 살짝 들여다보자. 앞으로 나오는 내용은 모두 Java 8 기준이다.

FileChannel.write(ByteBuffer)

HeapByteBuffer에 담겨 있는 내용을 파일에 저장하려면 대략 아래와 같은 코드를 사용하게 된다.

1
2
3
4
5
6
7
8
9
10
11

String contents = "abcde";
byte[] byteArr = contents.getBytes(StandardCharsets.UTF_8);

ByteBuffer byteBuffer = ByteBuffer.wrap(byteArr);  // HeapByteBuffer를 반환한다.
// 또는
// ByteBuffer byteBuffer = ByteBuffer.allocate(byteArr.length);    // HeapByteBuffer를 반환한다.
// byteBuffer.put(byteArr);
// byteBuffer.flip();

System.out.println("isDirect? " + byteBuffer.isDirect());  // false
fileChannel.write(byteBuffer);

위 코드에는 DirectByteBuffer가 전혀 나오지 않는다.

그런데 막상 실행해서 jcmd로 Native 메모리를 모니터링 해보면, DirectByteBuffer가 사용하는 Native 메모리를 나타내는 Internal 항목이 위 사용한 Buffer의 크기만큼 증가하는 것을 확인할 수 있다. 대략 다음과 같은 내용이 표시된다.

분명히 HeapByteBuffer를 사용했는데 왜 Native 메모리가 동원되는 걸까?

FileChannelImpl.write(ByteBuffer) 와 그 이후

특별히 지정하지 않는다면 인터페이스인 FileChannel의 구현체로 sun.nio.ch.FileChannelImpl이 사용된다. 코드는 아래와 같다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

// sun.nio.ch.FileChannelImpl

    public int write(ByteBuffer src) throws IOException {
        ensureOpen();
        if (!writable)
            throw new NonWritableChannelException();
        synchronized (positionLock) {
            int n = 0;
            int ti = -1;
            try {
                begin();
                ti = threads.add();
                if (!isOpen())
                    return 0;
                do {
                    n = IOUtil.write(fd, src, -1, nd);  //<=== 여기!!!
                } while ((n == IOStatus.INTERRUPTED) && isOpen());
                return IOStatus.normalize(n);
            } finally {
                threads.remove(ti);
                end(n > 0);
                assert IOStatus.check(n);
            }
        }
    }

IOUtil.write()는 다음과 같다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

// sun.nio.ch.IOUtil

    static int write(FileDescriptor fd, ByteBuffer src, long position,
                     NativeDispatcher nd)
        throws IOException
    {
        if (src instanceof DirectBuffer)
            return writeFromNativeBuffer(fd, src, position, nd);

        // Substitute a native buffer
        int pos = src.position();
        int lim = src.limit();
        assert (pos <= lim);
        int rem = (pos <= lim ? lim - pos : 0);
        ByteBuffer bb = Util.getTemporaryDirectBuffer(rem);  //<=== 여기!!!
        try {
            bb.put(src);
            bb.flip();
            // Do not update src until we see how many bytes were written
            src.position(pos);

            int n = writeFromNativeBuffer(fd, bb, position, nd);
            if (n > 0) {
                // now update src
                src.position(pos + n);
            }
            return n;
        } finally {
            Util.offerFirstTemporaryDirectBuffer(bb);
        }
    }

오호 특이한 점이 눈에 들어온다. 인자로 받아온 ByteBuffer의 Type이 DirectBuffer이면 writeFromNativeBuffer()를 호출하고 반환하지만, DirectBuffer가 아니면 Util.getTemporaryDirectBuffer(rem) 이렇게 슬그머니 DirectBuffer를 생성한다!!

잠시 곁다리로 빠져보자면, 몰래 대체품을 만들어 쓰기는 하지만 그래도 양심은 있는지 다음과 같이 개발자가 HeapByteBuffer 생성 시 지정한 크기가 아니라 실제 read/write 할 데이터 크기만큼의 DirectByteBuffer만 생성하는 점은 그래도 높이 쳐줄 수 있다. 그런데 이마저도 나중에 살펴볼 BufferCache 동작 방식을 생각해보면 좋다고만 할 수는 없다.

1
2

int rem = (pos <= lim ? lim - pos : 0);  // 버퍼의 크기가 아니라 실제 read/write 해야할 데이터 크기(limit - position)
ByteBuffer bb = Util.getTemporaryDirectBuffer(rem);  //<=== 여기!!!

이어서 계속 따라가보자. Util.getTemporaryDirectBuffer(rem)은 다음과 같다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

// sun.nio.ch.Util

    public static ByteBuffer getTemporaryDirectBuffer(int size) {
        // If a buffer of this size is too large for the cache, there
        // should not be a buffer in the cache that is at least as
        // large. So we'll just create a new one. Also, we don't have
        // to remove the buffer from the cache (as this method does
        // below) given that we won't put the new buffer in the cache.
        if (isBufferTooLarge(size)) {
            return ByteBuffer.allocateDirect(size);  //<=== 여기!!!
        }

        BufferCache cache = bufferCache.get();
        ByteBuffer buf = cache.get(size);
        if (buf != null) {
            return buf;
        } else {
            // No suitable buffer in the cache so we need to allocate a new
            // one. To avoid the cache growing then we remove the first
            // buffer from the cache and free it.
            if (!cache.isEmpty()) {
                buf = cache.removeFirst();
                free(buf);
            }
            return ByteBuffer.allocateDirect(size);  //<=== 여기!!!
        }
    }

대충 뭔가 재사용을 위해 캐시 개념을 사용하는 것 같은데 여튼 결국에는 ByteBuffer.allocateDirect(size)를 호출해서 DirectByteBuffer를 생성한다. ByteBuffer.allocateDirect(size)는 다음과 같다.

1
2
3
4
5

// java.nio.ByteBuffer

    public static ByteBuffer allocateDirect(int capacity) {
        return new DirectByteBuffer(capacity);
    }

즉, 특별한 설정 없이 일반적인 상황에서 FileChannel.write()을 사용하면 개발자가 작성한 프로그램 코드에서 HeapByteBuffer를 사용했더라도 내부적으로는 그 HeapByteBuffer가 사용되지 않고 항상 DirectByteBuffer가 사용된다. 코드 확인 결과 FileChannel.read()도 마찬가지다.

상당히 당황스럽다. 게다가 이런 얘기를 왜 API 문서나 기타 자료에서 쉽게 접할 수 없는지는 솔직히 의문이다.

암튼 그래.. 내가 쓰라고 한 건 무시하고 나 몰래 DirectByteBuffer를 만들어서 사용하네.. 근데 뭐가 문제임? 어쨌거나 잘 돌면 되는 거 아님?

이제부터 현장에서 발생했던 사례 얘기를 풀어본다. 만약 늘 아래 사례와 같이 동작한다면 상당히 심각한 버그라고 볼 수 있는데, 워낙 개발을 못 하는지라 어느 부분에선가 내가 코드를 잘못 짰을 수도 있기 때문에 늘 발생하는 상황이라고 단정할 수는 없다. 어쨌든 호기심이 생긴다면 이어서 쭉 보자.

DirectByteBuffer 메모리 회수

앞에서 DirectByteBuffer를 사용하면 DMA 혜택을 얻지만 메모리 관리 부담이 생긴다고 했다. DirectByteBuffer 메모리는 어떻게 회수되는 걸까? 여러 자료 찾아봤는데 대략 이런 말로 귀결된다.

Native 메모리를 참조하는 객체는 결국 JVM Heap 안에 생성되며,
이 객체가 JVM의 GC에 의해 회수되면 이 객체가 참조하는 Native 메모리는
JVM이 아닌 다른 메커니즘에 의해 어쨌든 회수된다.

요는 DirectByteBuffer를 사용해도 간접적이긴 하지만 결국에는 JVM GC에 의해 회수가 시작된다는 얘기다. 오 그럼 바로 회수되는 건 아니지만 다행스럽게도 결국 JVM GC가 챙겨주시는 거나 마찬가지네~

그런데 위 설명과는 다르게 어느 정도 시간이 지나면 결국 늘 이 분을 영접하게 되었다.

1

java.lang.OutOfMemoryError: Direct buffer memory

처음에는 ‘아 왜요~~ 저 DirectByteBuffer 쓰지도 않는데 저한테 왜 이러세요 진짜~‘ 였다. 그런데 에러 로그를 따라가보니 위에서 설명한 것처럼 나 몰래 응큼하게 내부적으로 DirectByteBuffer가 사용된다는 것까지는 알게 되었다. 그런데 메모리 반납은? 나 몰래 만들었으면 나 몰래 반납도 해줘야 하는 거 아님? 난 쪼렙 하수지만 넌 신성한 JDK 잖아!

여러 번 테스트 해봤는데 몇 시간, 심지어 며칠이 지나도 반납 안 해주더라.. JDK고 나발이고 나는 분명히 HeapByteBuffer를 전달해줬는데 나 몰래 DirectByteBuffer랑 바람 피우고, 그것도 모자라 지가 쓴 카드값까지 나한테..

어쨌든 상황을 정리해보면 다음과 같다.

일반적으로 FileChannel에 데이터를 write할 때는 결국 항상 DirectByteBuffer가 사용되는데,
OutOfMemoryError: Direct buffer memory가 계속 발생하는 걸로 봐서는,
DirectByteBuffer로 사용된 Native 메모리가 제대로 회수되지 않는(것 같)다.

자바에 내가 명시적으로 GC를 확실하게 유발할 수 있는 수단이 있는 것도 아닌데.. 망했.. 이러면 FileChannel은 못 쓰는 건데.. API 문서에도 다른 자료에도 왜 시원한 해법이 없지? 설마 아무도 FileChannel을 안 쓰는 건가? 그럴리가.. 내가 어딘가 잘못 짠 거겠지..

별 생각이 다 드는 가운데 여기서 대반전!

DirectByteBuffer 메모리 회수 방법

DirectByteBuffer 메모리는, JVM의 Heap 밖에 있어서 JVM GC가 아닌 다른 메커니즘에 의해 회수된다는 그 DirectByteBuffer 메모리는, 놀랍게도 Java 코드로 명시적으로 바로 회수할 수 있는 방법이 있었다. 아무리 검색해도 찾을 수가 없던 희귀한 내용이지만 바로 공유한다. 위에서 살펴본 코드 중에 답이 있었다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

// sun.nio.ch.Util

    public static ByteBuffer getTemporaryDirectBuffer(int size) {
        // If a buffer of this size is too large for the cache, there
        // should not be a buffer in the cache that is at least as
        // large. So we'll just create a new one. Also, we don't have
        // to remove the buffer from the cache (as this method does
        // below) given that we won't put the new buffer in the cache.
        if (isBufferTooLarge(size)) {
            return ByteBuffer.allocateDirect(size);
        }

        BufferCache cache = bufferCache.get();
        ByteBuffer buf = cache.get(size);
        if (buf != null) {
            return buf;
        } else {
            // No suitable buffer in the cache so we need to allocate a new
            // one. To avoid the cache growing then we remove the first
            // buffer from the cache and free it.
            if (!cache.isEmpty()) {
                buf = cache.removeFirst();
                free(buf);            //<=== 여기!!!
            }
            return ByteBuffer.allocateDirect(size);
        }
    }

엉? free(buf)가 있네? 어떻게 생겼나 한 번 볼까? 부왘ㅋㅋ 대박!!

1
2
3
4
5

// sun.nio.ch.Util

    private static void free(ByteBuffer buf) {
        ((DirectBuffer)buf).cleaner().clean();
    }

응큼하게 바람 피우고 카드값 떠넘기는 HeapByteBuffer와 결별(사실 HeapByteBuffer는 죄가 없다. FileChannel 구현부가 죄인이지)하고 내가 그냥 DirectByteBuffer와 사랑에 빠지기로 했다. 그래서 다음과 같이 HeapByteBuffer를 사용하던 코드를

1
2
3
4
5
6
7

ByteBuffer buf1 = ByteBuffer.allocate(size);
...
buf1.flip();
fileChannel.write(buf1);
...
ByteBuffer buf2 = ByteBuffer.wrap(byteArray);
fileChannel.write(buf2);

다음과 같이 명시적으로 DirectByteBuffer를 생성하고 사용하고 회수하도록 모두 바꾸고나니,

1
2
3
4
5
6
7
8
9
10
11

try {
  ByteBuffer directBuffer = ByteBuffer.allocateDirect(size);
  ...
  directBuffer.flip();
  fileChannel.write(directBuffer);
  ...
} catch (Exception e) {
  ...
} finally {
  ((DirectBuffer)directBuffer).cleaner().clean();
}

놀랍게도 ((DirectBuffer)directBuffer).cleaner().clean()가 호출된 후 jcmd로 확인해보면 Internal 영역이 명시적으로 사용한 DirectByteBuffer 크기만큼 바로 줄어드는 것을 확인할 수 있었다. 그리고 영 반갑지 않은 java.lang.OutOfMemoryError: Direct buffer memory도 다시 볼 일 없게 됐다.

이렇게 간단하고 직접적인 해결 방법이 이미 존재하는데 왜 그런 게 API 문서에 언급조차 없는지, 그리고 그 해결법도 왜 public static이 아니라 private static 메서드로 선언해둔 건지는 여전히 의문이다. 이 정도면 거의 일부러 감춰둔 정도 같기도 해서 ‘이거 써도 되는 거야?’라는 의문조차 들 정도..

그런데 한 가지 궁금한 게 더 있다.

HeapByteBuffer를 전달해줘도 내부적으로 응큼하게 DirectByteBuffer를 몰래 만드는 getTemporaryDirectBuffer() 내부에서, DirectByteBuffer 메모리를 회수할 수 있는 free() 메서드가 호출되고 있음에도 불구하고 몰래 만들어진 DirectByteBuffer가 회수되지 않는 이유는 뭘까? 이건 BufferCache를 보면 알 수 있다.

BufferCache

코드를 다시 보자.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

// sun.nio.ch.Util

    public static ByteBuffer getTemporaryDirectBuffer(int size) {
        // If a buffer of this size is too large for the cache, there
        // should not be a buffer in the cache that is at least as
        // large. So we'll just create a new one. Also, we don't have
        // to remove the buffer from the cache (as this method does
        // below) given that we won't put the new buffer in the cache.
        if (isBufferTooLarge(size)) {
            return ByteBuffer.allocateDirect(size);
        }

        BufferCache cache = bufferCache.get();
        ByteBuffer buf = cache.get(size);
        if (buf != null) {
            return buf;
        } else {
            // No suitable buffer in the cache so we need to allocate a new
            // one. To avoid the cache growing then we remove the first
            // buffer from the cache and free it.
            if (!cache.isEmpty()) {
                buf = cache.removeFirst();  //<=== 여기!!!
                free(buf);            //<=== 여기!!!
            }
            return ByteBuffer.allocateDirect(size);
        }
    }

BufferCache에 동일한 크기의 DirectByteBuffer가 있으면 그걸 재사용하고, 없으면 캐시에 있는 다른 크기의 DirectByteBuffer를 free()를 이용해서 하나 삭제한다. 그렇게 해서 캐쉬의 총 갯수가 늘어나지 않게 한다. 실제로도 이렇게 잘 동작한다.

예를 들어 크기가 10M로 모두 같은 HeapByteBuffer를 3개 생성해서 FileChannel.write()에 사용하면 내부적으로 DirectByteBuffer가 생성되므로 10M 짜리 DirectByteBuffer 3개, 총 30M가 사용될 것 같지만, 위에 나오는 BufferCache 덕분에 실제로는 10M 짜리 DirectByteBuffer 한 개만 생성되고 재사용된다.

여기까지는 좋은데 문제는 맨 마지막 부분 return ByteBuffer.allocateDirect(size)에서 새로 생성된 후 반환되는 DirectByteBuffer는 회수되지 않는(걸로 보인)다는 점이다.

BufferCache는 아래와 같이 ThreadLocal에 담겨서 per-thread로 존재한다.

1
2
3
4
5
6
7
8
9
10
11

// sun.nio.ch.Util

    // Per-thread cache of temporary direct buffers
    private static ThreadLocal<BufferCache> bufferCache =
        new ThreadLocal<BufferCache>()
    {
        @Override
        protected BufferCache initialValue() {
            return new BufferCache();
        }
    };

위에서 개발자가 명시적으로 생성한 HeapByteBuffer 대신 내부적으로 DirectByteBuffer를 생성할 때 실제 read/write 할 만큼의 DirectByteBuffer를 생성한다고 했다. 필요한 만큼만 새로 생성하므로 메모리 사용량에 있어서는 유리하지만, 그 필요한 만큼이 그때그때 다른 상황에서는 지금 살펴본 BufferCache의 hit율이 떨어져서 DirectByteBuffer의 생성 빈도가 많아질 수 있다. ByteBuffer API 문서에 따르면 DirectByteBuffer는 메모리 할당/해제 비용이 HeapByteBuffer보다 더 크다고 한다. 따라서 DirectByteBuffer 생성 빈도가 많으면 성능에 악영향을 미칠 수 있다.

정리하면, 하나의 스레드에서 이런 비명시적 방식(개발자가 HeapByteBuffer를 FileChannel.write()에 인자로 전달해줘도 FileChannelImpl이 내부적으로 몰래 DirectByteBuffer를 생성하는 방식)으로 DirectByteBuffer가 생성되면,

• 크기가 동일한 HeapByteBuffer를 여러개 만들어도
• BufferCache 덕분에 해당 스레드 내에서는 DirectByteBuffer가 하나만 만들어지고 재사용될 수는 있지만,
• 그 한 개의 DirectByteBuffer가 제대로 회수되지 않으면,
• 새로운 스레드가 실행될 때마다 계속 누적되다가 결국 OutOfMemory 에러를 맞이하게 된다.

XX:MaxDirectMemorySize 옵션

DirectByteBuffer가 사용하는 Native Memory 최대 크기를 지정할 수 있는 옵션이 있다. XX:MaxDirectMemorySize인데 DirectByteBuffer 메모리가 제대로 회수되지 않고 누적되다가 최대 크기를 넘는다면 어떻게 될까?

1. 오랫동안 사용되지 않고 메모리만 점유하고 있던 DirectByteBuffer를 알아서 회수한다.
2. java.lang.OutOfMemoryError: Direct buffer memory가 발생한다.

혹시 하는 마음에 1을 기대했는데, 현실은 2다.

실제 검증 - 몰래 만들어진 DirectByteBuffer 메모리도 회수된다!!

내가 잘못 짰을 수도 있는 로직과 뒤섞인 상태로는, 몰래 만들어지는 DirectByteBuffer가 항상 회수되지 않는다고 확언을 할 수 없으므로 실험용 간단한 프로그램을 만들어서 검증해봤다.

https://github.com/HomoEfficio/scratchpad-bytebuffer 를 참고하면 직접 요리조리 해볼 수 있다.

README에 있는대로, 프로그램 실행해서 VisualVM 으로 강제 GC를 시킨 후에 jcmd로 확인해보면 Native Memory가 회수되는 것을 확인할 수 있었다.

따라서 몰래 만들어진 DirectByteBuffer도 정상적인 경우라면 해당 DirectByteBuffer를 참조하는 객체(A라고 하자)가 GC될 때 DirectByteBuffer 메모리도 함께 회수된다. 라고 결론 지을 수 있겠다.

하지만 그래도 A가 언제 GC 될 지 알 수 없고, GC 전까지 DirectByteBuffer는 계속 Native Memory를 점유하게 된다. 아마도 잘못 작성한 코드 때문이겠지만 알 수 없는 이유로 Native Memory 가 장시간 계속 회수되지 않는다면, 우리에겐 ((DirectBuffer)directBuffer).cleaner().clean() 라는 무기가 있다는 사실을 알아두면 좋다.

마무리

• 특별한 FileChannel 구현체를 사용하지 않는다면, FileChannel에 read/write 할 때 HeapByteBuffer를 사용해도 내부적으로 DirectByteBuffer가 사용된다.

• 내부적으로 사용되는 DirectByteBuffer가 제대로 회수되지 않으면 java.lang.OutOfMemoryError: Direct buffer memory가 발생할 수 있다.

• 이렇게 FileChannel에 read/write 하는 코드에 의해 OOM이 발생한다면,

  • HeapByteBuffer을 사용하지 말고 명시적으로 DirectByteBuffer를 사용하고,
  • ((DirectBuffer)directBuffer).cleaner().clean()를 사용해서 명시적으로 해제하자.

함께 보기

• Java NIO는 생각만큼 non-blocking 하지 않다
• Java NIO Direct Buffer를 이용해서 대용량 파일 행 기준으로 쪼개기
• 대용량 파일을 AsynchronousFileChannel로 다뤄보기
• Java Native Memory Tracking

WebFlux 사용 시 Controller 단에서 RequestBody 를 인자로 받을 때,

다음과 같이 Mono를 받아오도록 작성해야할까?

1
2
3
4

@PostMapping("/mono")
public Mono<SellerOut> createWithMono(@RequestBody Mono<SellerIn> sellerIn) {
    return sellerService.createWithMono(sellerIn);
}

아니면 그냥 Raw 객체를 받아오도록 작성해야할까?

1
2
3
4

@PostMapping("/entity")
public Mono<SellerOut> createWithRaw(@RequestBody SellerIn sellerIn) {
    return sellerService.createWithRaw(sellerIn);
}

non-blocking 드라이버를 제공해주는 MongoDB를 사용한다고 가정하고 사용성과 성능 관점에서 살펴보자.

사용성

컨트롤러 단에서는 위에서 보는 것처럼 큰 차이가 없다. 서비스가 호출하는 데이터 접근 계층(Data Access Layer)에서 작지만 큰 차이가 발생한다.

ReactiveCrudRepository

스프링 WebFlux 환경에서도 사용하기 쉽게 추상화 된 ReactiveCrudRepository를 통해 쉽게 데이터 저장소에 접근할 수 있다. MongoDB 용으로 특화된 ReactiveMongoRepository를 사용하면 Spring Data 에서 제공해주는 편리한 메서드 이름 조합 방식을 WebFlux에서도 사용할 수 있다. 이는 사용성과 생산성을 높여주는 데 큰 역할을 한다.

그런데 엔티티를 저장할 때 사용되는 save() 메서드는 다음과 같이 Mono가 아니라 Raw 객체를 사용하도록 정의돼 있다.

1
2
3
4
5
6
7
8
9
10
11
12

@NoRepositoryBean
public interface ReactiveCrudRepository<T, ID> extends Repository<T, ID> {

  /**
   * Saves a given entity. Use the returned instance for further operations as the save operation might have changed the
   * entity instance completely.
   *
   * @param entity must not be {@literal null}.
   * @return {@link Mono} emitting the saved entity.
   * @throws IllegalArgumentException in case the given {@literal entity} is {@literal null}.
   */
  <S extends T> Mono<S> save(S entity);  // Mono가 아니라 걍 S